defmodule Aerospike do @moduledoc """ Public entry point for the Aerospike Elixir driver. `Aerospike` exposes the low-level API for starting a supervised cluster, constructing keys, and running commands against that cluster. Most applications should define an `Aerospike.Repo` module to bind this API to one configured cluster; lower-level code can call this module directly when it needs explicit control over the cluster name. The main command families are: * `get/3` reads all bins for a key * `get_header/2` reads only record metadata for a key * `put/4` writes a bin map * `put_payload/4` sends a caller-built single-record write/delete frame * `apply_udf/6` executes one record UDF against one key * `register_udf/3`, `register_udf/4`, `remove_udf/2`, `remove_udf/3`, and `list_udfs/1` / `list_udfs/2` manage server-side UDF packages and return explicit metadata/task handles * `truncate/2`, `truncate/3`, and `truncate/4` expose one-node operator truncation helpers with explicit namespace and set forms * `create_user/5`, `create_pki_user/4`, `drop_user/3`, `change_password/4`, `grant_roles/4`, `revoke_roles/4`, `query_user/3`, `query_users/2`, `create_role/4`, `drop_role/3`, `set_whitelist/4`, `set_quotas/5`, `grant_privileges/4`, `revoke_privileges/4`, `query_role/3`, and `query_roles/2` expose the enterprise security-admin seam * `add/4`, `append/4`, and `prepend/4` expose thin unary write helpers for common counter and string mutations * `metrics_enabled?/1`, `enable_metrics/2`, `disable_metrics/1`, `stats/1`, and `warm_up/2` expose opt-in runtime metrics and an explicit operator pool probe over the already-started workers * `touch/2` updates record metadata * `delete/2` removes a record * `exists/2` performs a header-only existence probe * `operate/4` runs simple, CDT-style, and expression operation lists built with `Aerospike.Op`, `Aerospike.Op.List`, `Aerospike.Op.Map`, `Aerospike.Op.Exp`, and `Aerospike.Ctx` * unary commands (`get/3`, `get_header/2`, `put/4`, `exists/2`, `touch/2`, `delete/2`, `operate/4`, `apply_udf/6`, `add/4`, `append/4`, and `prepend/4`) accept `%Aerospike.Exp{}` via `:filter` for server-side execution filtering * `Aerospike.Exp` builds server-side expression values, including values usable as expression-backed secondary-index sources * `create_expression_index/5` creates expression-backed secondary indexes on servers that support them and returns a pollable index task * `batch_get/4`, `batch_get_header/3`, `batch_exists/3`, `batch_get_operate/4`, `batch_delete/3`, and `batch_udf/6` operate on multiple keys and return per-key results in caller order * `Aerospike.Batch` and `batch_operate/3` expose a curated heterogeneous batch surface for mixing reads, writes, deletes, operations, and record UDF calls while returning one `%Aerospike.BatchResult{}` per input * `child_spec/1`, `close/2`, `key/3`, and `key_digest/3` round out the root lifecycle and key-construction boundary * `info/3`, `info_node/4`, `nodes/1`, and `node_names/1` expose one-node operator reads over the published cluster view; `info_node/4` targets one named active node discovered from `node_names/1` or `nodes/1` * `set_xdr_filter/4` sets or clears Enterprise XDR expression filters through a one-node info command * `query_stream!/3`, `query_all/3`, `query_count/3`, `query_aggregate/6`, and `query_aggregate_result/6` run secondary-index queries through the same node-preparation pipeline, with lazy outer streams but node-buffered record delivery and optional `%Aerospike.Exp{}` filters through `Query.filter/2` when a secondary-index predicate from `Query.where/2` is also used. * `query_execute/4` and `query_udf/6` run background query jobs on that same setup path and return pollable task handles * `scan_stream/3`, `scan_stream!/3`, `scan_all/3`, `scan_all!/3`, `scan_count/3`, `scan_count!/3`, `scan_page/3`, and `scan_page!/3` run scan fan-out across the same scan/runtime setup, again with lazy outer streams, node-buffered record delivery, and resumable pages * scan/query helpers that already support node targeting accept `node: node_name` in `opts`; discover names with `node_names/1` or `nodes/1` Quick-start shape: {:ok, _sup} = Aerospike.start_link( name: :aerospike, transport: Aerospike.Transport.Tcp, hosts: ["127.0.0.1:3000"], namespaces: ["test"], pool_size: 2 ) Aerospike.Cluster.ready?(:aerospike) Scan and query stream helpers are lazy at the `Enumerable` boundary. The runtime buffers each node's response before yielding that node's records downstream. Expression builders cover scalar, metadata, arithmetic, conditional, variable, CDT, bit, and HyperLogLog helper families. Public command and startup options are keyword-based and validated by the facade before command execution. """ alias Aerospike.Batch alias Aerospike.Cluster alias Aerospike.Cluster.Supervisor, as: ClusterSupervisor alias Aerospike.Command.Admin alias Aerospike.Command.ApplyUdf alias Aerospike.Command.Batch, as: MixedBatch alias Aerospike.Command.BatchCommand.Entry alias Aerospike.Command.BatchDelete alias Aerospike.Command.BatchGet alias Aerospike.Command.BatchUdf alias Aerospike.Command.Delete alias Aerospike.Command.Exists alias Aerospike.Command.Get alias Aerospike.Command.Operate alias Aerospike.Command.Put alias Aerospike.Command.PutPayload alias Aerospike.Command.ScanOps alias Aerospike.Command.Touch alias Aerospike.Command.WriteOp alias Aerospike.Error alias Aerospike.ExecuteTask alias Aerospike.Exp alias Aerospike.IndexTask alias Aerospike.Key alias Aerospike.Page alias Aerospike.Policy alias Aerospike.Privilege alias Aerospike.Protocol.AsmMsg.Operation alias Aerospike.Protocol.OperateFlags alias Aerospike.Query alias Aerospike.RegisterTask alias Aerospike.RetryPolicy alias Aerospike.Role alias Aerospike.Runtime.TxnRoll alias Aerospike.RuntimeMetrics alias Aerospike.Scan alias Aerospike.Txn alias Aerospike.UDF alias Aerospike.User @typedoc """ Identifier for a running cluster facade. Read-side helpers accept the registered cluster name or a pid registered under that name. Arbitrary `GenServer.server()` forms are not supported. """ @type cluster :: named_cluster() | pid() @typedoc """ Registered atom name for a running cluster. Lifecycle and transaction helpers resolve supervisor and ETS resources from this name, so they do not currently accept arbitrary `GenServer.server()` identities. """ @type named_cluster :: atom() @typedoc """ One active cluster node as returned by `nodes/1`. """ @type node_info :: %{name: String.t(), host: String.t(), port: :inet.port_number()} @typedoc "Security user metadata returned by `query_user/3` and `query_users/2`." @type user_info :: User.t() @typedoc "Security role metadata returned by `query_role/3` and `query_roles/2`." @type role_info :: Role.t() @typedoc "Security privilege metadata used by role administration APIs." @type privilege :: Privilege.t() @typedoc """ Non-negative timeout value in milliseconds. `:timeout` is the total client-side budget for a command. `:socket_timeout` is the per-attempt idle socket deadline; `0` asks the command path to use the remaining total budget. """ @type timeout_ms :: non_neg_integer() @typedoc """ Record TTL accepted by write commands. * non-negative integer - explicit TTL in seconds * `:default` - use the namespace default TTL * `:never_expire` - ask the server to keep the record indefinitely * `:dont_update` - preserve the existing record TTL """ @type ttl :: non_neg_integer() | :default | :never_expire | :dont_update @typedoc """ AP namespace read consistency. `:one` reads from one available replica. `:all` asks the server to consult all relevant replicas for the read. """ @type read_mode_ap :: :one | :all @typedoc """ Strong-consistency namespace read consistency. The default is `:session`. `:linearize` requests linearized reads. `:allow_replica` and `:allow_unavailable` relax consistency for availability when the server namespace configuration allows it. """ @type read_mode_sc :: :session | :linearize | :allow_replica | :allow_unavailable @typedoc """ Record existence behavior for write commands. * `:update` - create or update bins (default) * `:update_only` - update an existing record only * `:create_or_replace` - create a new record or replace the existing record * `:replace_only` - replace an existing record only * `:create_only` - create only when the key is absent """ @type record_exists_action :: :update | :update_only | :create_or_replace | :replace_only | :create_only @typedoc """ Generation check behavior for write commands. `:none` disables generation checks. `:expect_equal` requires the server record generation to equal `:generation`. `:expect_gt` requires it to be greater than `:generation`. When `:generation` is positive and `:generation_policy` is omitted, the driver defaults to `:expect_equal`. """ @type generation_policy :: :none | :expect_equal | :expect_gt @typedoc """ Commit acknowledgement level for write commands. `:all` waits for the server's configured replica commit level. `:master` accepts acknowledgement from the master partition owner. """ @type commit_level :: :all | :master @typedoc """ Retry option accepted by single-record, scan, and query command families. * `:max_retries` - retries after the first attempt * `:sleep_between_retries_ms` - fixed delay between retry attempts * `:replica_policy` - read retry routing policy, `:master` or `:sequence` Retry defaults are configured at cluster start and can be overridden per command where this type appears. """ @type retry_opt :: {:max_retries, non_neg_integer()} | {:sleep_between_retries_ms, non_neg_integer()} | {:replica_policy, RetryPolicy.replica_policy()} @typedoc """ Read option accepted by `get/4`, `get_header/3`, and `exists/3`. Read helpers accept timeout/retry opts, consistency opts, `:send_key`, `:use_compression`, and a server-side expression `:filter`. """ @type read_opt :: {:timeout, timeout_ms()} | {:socket_timeout, timeout_ms()} | retry_opt() | {:read_mode_ap, read_mode_ap()} | {:read_mode_sc, read_mode_sc()} | {:read_touch_ttl_percent, -1 | 0..100} | {:send_key, boolean()} | {:use_compression, boolean() | nil} | {:filter, Exp.t() | nil} @typedoc "Keyword options accepted by read helpers." @type read_opts :: [read_opt()] @typedoc """ Write option accepted by single-record write, delete, UDF, and operate helpers. Write helpers accept timeout/retry opts, record TTL and generation policy, existence policy, commit level, durable delete, `:respond_per_op`, `:send_key`, read-touch metadata opts used by mixed operations, per-command compression, a server-side expression `:filter`, and an optional transaction handle. """ @type write_opt :: {:timeout, timeout_ms()} | {:socket_timeout, timeout_ms()} | retry_opt() | {:ttl, ttl()} | {:generation, non_neg_integer()} | {:generation_policy, generation_policy()} | {:exists, record_exists_action()} | {:commit_level, commit_level()} | {:durable_delete, boolean()} | {:respond_per_op, boolean()} | {:send_key, boolean()} | {:read_mode_ap, read_mode_ap()} | {:read_mode_sc, read_mode_sc()} | {:read_touch_ttl_percent, -1 | 0..100} | {:use_compression, boolean() | nil} | {:filter, Exp.t() | nil} | {:txn, Txn.t()} @typedoc "Keyword options accepted by single-record write, delete, UDF, and operate helpers." @type write_opts :: [write_opt()] @typedoc """ Parent batch dispatch option. Parent options describe the batch request as a whole: deadline, node concurrency, partial-result handling, response shape, and inline execution hints. Per-record read/write policy belongs on `batch_record_read_opt/0` and `batch_record_write_opt/0`. """ @type batch_parent_opt :: {:timeout, timeout_ms()} | {:socket_timeout, timeout_ms()} | {:max_concurrent_nodes, non_neg_integer()} | {:allow_partial_results, boolean()} | {:respond_all_keys, boolean()} | {:allow_inline, boolean()} | {:allow_inline_ssd, boolean()} @typedoc "Parent batch options accepted by `batch_operate/3`." @type batch_parent_opts :: [batch_parent_opt()] @typedoc """ Batch read option accepted by per-entry batch reads. These options are encoded with one batch read entry. Parent dispatch options are only accepted by homogeneous batch read helpers and `batch_operate/3`. """ @type batch_record_read_opt :: {:filter, Exp.t() | nil} | {:read_mode_ap, read_mode_ap()} | {:read_mode_sc, read_mode_sc()} | {:read_touch_ttl_percent, -1 | 0..100} @typedoc "Keyword options accepted by per-entry batch reads." @type batch_record_read_opts :: [batch_record_read_opt()] @typedoc "Batch read option accepted by homogeneous batch-read helpers." @type batch_read_opt :: batch_parent_opt() | batch_record_read_opt() @typedoc "Keyword options accepted by batch read helpers." @type batch_read_opts :: [batch_read_opt()] @typedoc """ Batch write option accepted by per-entry batch writes. These options are encoded with one batch write/delete/UDF entry. Parent dispatch options are only accepted by homogeneous batch write helpers and `batch_operate/3`. """ @type batch_record_write_opt :: {:ttl, ttl()} | {:generation, non_neg_integer()} | {:generation_policy, generation_policy()} | {:exists, record_exists_action()} | {:commit_level, commit_level()} | {:durable_delete, boolean()} | {:respond_per_op, boolean()} | {:send_key, boolean()} | {:filter, Exp.t() | nil} | {:read_mode_ap, read_mode_ap()} | {:read_mode_sc, read_mode_sc()} | {:read_touch_ttl_percent, -1 | 0..100} @typedoc "Keyword options accepted by per-entry batch writes." @type batch_record_write_opts :: [batch_record_write_opt()] @typedoc "Batch write option accepted by homogeneous batch-write helpers." @type batch_write_opt :: batch_parent_opt() | batch_record_write_opt() @typedoc "Keyword options accepted by batch write helpers." @type batch_write_opts :: [batch_write_opt()] @typedoc "Node-targeting option accepted by scan/query helpers that support single-node execution." @type node_opt :: {:node, String.t()} @typedoc """ Runtime option accepted by all scan and query helpers. These options tune client runtime behavior and server scan/query fields: command deadlines, retries, task timeout, pool checkout timeout, fan-out concurrency, throttling, bin-data inclusion, expected duration, task id, and cursor resume state. """ @type scan_query_runtime_opt :: {:timeout, timeout_ms()} | {:socket_timeout, timeout_ms()} | retry_opt() | {:task_timeout, timeout_ms() | :infinity} | {:pool_checkout_timeout, timeout_ms()} | {:max_concurrent_nodes, non_neg_integer()} | {:records_per_second, non_neg_integer()} | {:include_bin_data, boolean()} | {:expected_duration, :long | :short | :long_relax_ap} | {:task_id, pos_integer()} | {:cursor, Aerospike.Cursor.t()} @typedoc """ Runtime option accepted by scan/query helpers that may target one node. `:node` must be an active node name returned by `node_names/1` or `nodes/1`. Final aggregate reduction does not accept `:node` because it must consume all server partials required to produce one local result. """ @type scan_query_opt :: scan_query_runtime_opt() | node_opt() @typedoc "Keyword options accepted by scan and query helpers." @type scan_query_opts :: [scan_query_opt()] @typedoc """ Additional local-source option for finalized aggregate queries. Exactly one of these is required by `query_aggregate_result/6`: inline Lua source via `:source`, or a readable local file path via `:source_path`. """ @type aggregate_result_opt :: {:source, String.t()} | {:source_path, Path.t()} @typedoc "Keyword options accepted by `query_aggregate_result/6` and `query_aggregate_result!/6`." @type aggregate_result_opts :: [scan_query_runtime_opt() | aggregate_result_opt()] @typedoc """ Option accepted by one-node info/admin helpers. `:pool_checkout_timeout` bounds checkout from the selected node pool before the one-node info/admin command is sent. """ @type admin_opt :: {:pool_checkout_timeout, timeout_ms()} @typedoc "Keyword options accepted by one-node info/admin helpers." @type admin_opts :: [admin_opt()] @typedoc """ Option accepted by security-admin helpers. `:timeout` bounds the admin protocol operation. `:pool_checkout_timeout` bounds checkout from the selected node pool before the command is sent. """ @type security_admin_opt :: {:timeout, timeout_ms()} | {:pool_checkout_timeout, timeout_ms()} @typedoc "Keyword options accepted by security-admin helpers." @type security_admin_opts :: [security_admin_opt()] @typedoc "Secondary-index collection type." @type index_collection :: :list | :mapkeys | :mapvalues @typedoc "Secondary-index particle/source type." @type index_type :: :numeric | :string | :geo2dsphere @typedoc """ Option accepted by `create_index/4`. `:bin`, `:name`, and `:type` are required. `:collection` creates a collection index over list values, map keys, or map values. `:ctx` targets a nested CDT path built with `Aerospike.Ctx`. """ @type create_index_opt :: {:bin, String.t()} | {:name, String.t()} | {:type, index_type()} | {:collection, index_collection()} | {:ctx, [Aerospike.Ctx.step()]} | admin_opt() @typedoc "Keyword options accepted by `create_index/4`." @type create_index_opts :: [create_index_opt()] @typedoc """ Option accepted by `create_expression_index/5`. `:name` and `:type` are required. Expression indexes use the provided `%Aerospike.Exp{}` as their source and therefore do not accept `:bin` or nested CDT `:ctx`. """ @type create_expression_index_opt :: {:name, String.t()} | {:type, index_type()} | {:collection, index_collection()} | admin_opt() @typedoc "Keyword options accepted by `create_expression_index/5`." @type create_expression_index_opts :: [create_expression_index_opt()] @typedoc """ Option accepted by `truncate/3` and `truncate/4`. `:before` limits truncation to records last updated before the given timestamp. `:pool_checkout_timeout` bounds checkout for the admin request. """ @type truncate_opt :: {:before, DateTime.t()} | admin_opt() @typedoc "Keyword options accepted by truncate helpers." @type truncate_opts :: [truncate_opt()] @typedoc "Option accepted by `enable_metrics/2`." @type metrics_opt :: {:reset, boolean()} @typedoc "Keyword options accepted by `enable_metrics/2`." @type metrics_opts :: [metrics_opt()] @typedoc """ Option accepted by `warm_up/2`. `:count` is the number of worker checkout probes per active node. `0` means up to the configured pool size. `:pool_checkout_timeout` bounds each probe. """ @type warm_up_opt :: {:count, non_neg_integer()} | {:pool_checkout_timeout, timeout_ms()} @typedoc "Keyword options accepted by `warm_up/2`." @type warm_up_opts :: [warm_up_opt()] @typedoc "Role-creation option accepted by `create_role/4`." @type create_role_opt :: {:whitelist, [String.t()]} | {:read_quota, non_neg_integer()} | {:write_quota, non_neg_integer()} | security_admin_opt() @typedoc "Keyword options accepted by `create_role/4`." @type create_role_opts :: [create_role_opt()] @typedoc "Transaction options accepted by `transaction/3` when a handle is not supplied." @type transaction_opts :: [{:timeout, timeout_ms()}] @typedoc "Operation input accepted by `operate/4`." @type operate_operation :: {:write, String.t() | atom(), term()} | {:read, String.t() | atom()} | {:add, String.t() | atom(), integer() | float()} | {:append, String.t() | atom(), String.t()} | {:prepend, String.t() | atom(), String.t()} | :touch | :delete | Aerospike.Op.t() @doc """ Starts a supervised cluster. Internally this delegates to `Aerospike.Cluster.Supervisor.start_link/1`, which boots the cluster-state owner, per-node pool supervisor, partition-map writer, and tend-cycle process under one `rest_for_one` tree. The `:name` option is the public cluster identity later passed to `get/3`, `Aerospike.Cluster.ready?/1`, and the other facade/read-side helpers. Required options: `:name`, `:transport`, `:hosts`, `:namespaces`. Startup validation happens synchronously at this boundary. Shape errors for required opts, retry/breaker knobs, TLS/connect opts, and auth pairs fail `start_link/1` immediately instead of surfacing later from the first tend cycle or pool worker. The `:name` remains the public cluster identity for facade calls such as `get/3`, `info/3`, and `Aerospike.Cluster.ready?/1`. Cluster lifecycle knobs: * `:tend_trigger` — `:timer` (default) or `:manual`. * `:tend_interval_ms` — automatic tend period in milliseconds. Positive integer. * `:failure_threshold` — consecutive tend failures before the tend cycle demotes a node. Non-negative integer. Pool-level knobs (forwarded internally on each node-pool start): * `:pool_size` — workers per node. Positive integer. * `:min_connections_per_node` — minimum warm connection target per node. Non-negative integer. `0` allows lazy worker creation. * `:idle_timeout_ms` — milliseconds a worker may sit idle before the pool verification step evicts it. Positive integer. Defaults stay under Aerospike's `proto-fd-idle-ms`. * `:max_idle_pings` — bound on how many idle workers NimblePool may drop per verification cycle. Positive integer. Breaker and retry knobs: * `:circuit_open_threshold` — consecutive node failures tolerated before new commands are refused. Non-negative integer. * `:max_concurrent_ops_per_node` — in-flight plus queued command cap enforced per node. Positive integer. * `:max_retries` — retries after the initial attempt. Non-negative integer. * `:sleep_between_retries_ms` — fixed delay between retries. Non-negative integer. * `:replica_policy` — `:master` or `:sequence`. Cluster feature toggles: * `:use_compression` — boolean cluster-wide request-compression opt-in, gated per node by advertised capabilities. * `:use_services_alternate` — boolean toggle for `peers-clear-alt` discovery. * `:seed_only_cluster` — boolean. When true, discovery is limited to configured seed addresses. * `:cluster_name` — expected server cluster name. Nodes reporting a different name are rejected. * `:application_id` — application identity sent to capable servers for server-side client correlation. Auth opts: * `:auth_mode` — `:internal` (default), `:external`, or `:pki`. External and PKI modes require `Aerospike.Transport.Tls`. * `:user` / `:password` — cluster-wide credentials. Must be passed together or omitted together. PKI auth uses the TLS client certificate and must omit both. * `:login_timeout_ms` — startup/login read deadline in milliseconds. TCP-level tuning knobs (passed verbatim to the TCP transport via the `:connect_opts` keyword): * `:connect_timeout_ms` — handshake + write-buffer drain deadline. * `:info_timeout` — read deadline applied to every `info/2` call. Defaults to `:connect_timeout_ms`. * `:tcp_nodelay` — boolean, default `true`. * `:tcp_keepalive` — boolean, default `true`. * `:tcp_sndbuf` / `:tcp_rcvbuf` — positive integer kernel buffer sizes. Unset lets the kernel pick. `Aerospike.Cluster.Supervisor` documents the underlying validation and child ownership details. """ @spec start_link([ClusterSupervisor.option()]) :: Supervisor.on_start() def start_link(opts) when is_list(opts) do ClusterSupervisor.start_link(opts) end @doc """ Returns a child specification for one supervised cluster. This delegates to the same cluster-supervisor implementation as `start_link/1`, so the accepted options and validation boundary match. """ @spec child_spec([ClusterSupervisor.option()]) :: Supervisor.child_spec() def child_spec(opts) when is_list(opts) do ClusterSupervisor.child_spec(opts) end @doc """ Stops the supervised cluster registered under the atom `cluster`. This targets the registered cluster supervisor name derived from `cluster` and returns `:ok` when the supervisor exits or is already absent. """ @spec close(named_cluster(), timeout :: non_neg_integer()) :: :ok def close(cluster, timeout \\ 15_000) when is_atom(cluster) and is_integer(timeout) and timeout >= 0 do case Process.whereis(ClusterSupervisor.sup_name(cluster)) do nil -> :ok pid -> Supervisor.stop(pid, :normal, timeout) end end @doc """ Builds a key from namespace, set, and a user key. This is a thin wrapper over `Aerospike.Key.new/3`. """ @spec key(String.t(), String.t(), String.t() | integer()) :: Key.t() def key(namespace, set, user_key), do: Key.new(namespace, set, user_key) @doc """ Builds a key from namespace, set, and an existing 20-byte digest. This is a thin wrapper over `Aerospike.Key.from_digest/3`. """ @spec key_digest(String.t(), String.t(), <<_::160>>) :: Key.t() def key_digest(namespace, set, digest), do: Key.from_digest(namespace, set, digest) @doc """ Reads `key` from `cluster`. Pass `:all` to read every bin, or pass a non-empty list of string or atom bin names to project only those bins. Options: * `:timeout` — total op-budget milliseconds for the call, shared across the initial send and every retry. Default `5_000`. * `:socket_timeout` — per-attempt socket idle deadline in milliseconds, capped by the remaining total budget. `0` uses the remaining total budget. * `:max_retries` — overrides the cluster-default retry cap for this call. `0` disables retry entirely. See `Aerospike.RetryPolicy`. * `:sleep_between_retries_ms` — fixed delay between retry attempts. * `:replica_policy` — `:master` (all attempts against the master) or `:sequence` (walk the replica list by attempt index). * `:read_mode_ap` — `:one` or `:all`. * `:read_mode_sc` — `:session`, `:linearize`, `:allow_replica`, or `:allow_unavailable`. * `:read_touch_ttl_percent` — `-1`, `0`, or an integer from `1` through `100`. * `:send_key` — include the user key in the request when `true`. * `:use_compression` — per-command compression override; `nil` keeps the cluster/node default. * `:filter` — non-empty `%Aerospike.Exp{}` server-side filter expression, or `nil` for no filter. Returns `{:ok, %Aerospike.Record{}}` on hit, `{:error, %Aerospike.Error{code: :key_not_found}}` on miss, or a routing atom (`:cluster_not_ready`, `:no_master`, `:unknown_node`) when the cluster view cannot serve the request. Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`. """ @spec get(cluster(), Key.key_input(), :all | [String.t() | atom()], read_opts()) :: {:ok, Aerospike.Record.t()} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready | :no_master | :unknown_node} def get(cluster, key, bins \\ :all, opts \\ []) def get(cluster, key, bins, opts) do with {:ok, key} <- coerce_key(key) do Get.execute(cluster, key, bins, opts) end end @doc """ Reads only record metadata for `key` from `cluster`. This is the explicit single-record header helper. It reuses the same unary read path as `get/3`, but requests only generation/TTL metadata and returns a `%Aerospike.Record{}` with `bins: %{}` on hit. Options: * `:timeout` — total op-budget milliseconds for the call, shared across the initial send and every retry. Default `5_000`. * `:socket_timeout` — per-attempt socket idle deadline in milliseconds, capped by the remaining total budget. `0` uses the remaining total budget. * `:max_retries` — overrides the cluster-default retry cap for this call. `0` disables retry entirely. See `Aerospike.RetryPolicy`. * `:sleep_between_retries_ms` — fixed delay between retry attempts. * `:replica_policy` — `:master` (all attempts against the master) or `:sequence` (walk the replica list by attempt index). * `:read_mode_ap` — `:one` or `:all`. * `:read_mode_sc` — `:session`, `:linearize`, `:allow_replica`, or `:allow_unavailable`. * `:read_touch_ttl_percent` — `-1`, `0`, or an integer from `1` through `100`. * `:send_key` — include the user key in the request when `true`. * `:use_compression` — per-command compression override; `nil` keeps the cluster/node default. * `:filter` — non-empty `%Aerospike.Exp{}` server-side filter expression, or `nil` for no filter. Returns `{:ok, %Aerospike.Record{bins: %{}}}` on hit, `{:error, %Aerospike.Error{code: :key_not_found}}` on miss, or a routing atom (`:cluster_not_ready`, `:no_master`, `:unknown_node`) when the cluster view cannot serve the request. Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`. """ @spec get_header(cluster(), Key.key_input(), read_opts()) :: {:ok, Aerospike.Record.t()} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready | :no_master | :unknown_node} def get_header(cluster, key, opts \\ []) def get_header(cluster, key, opts) when is_list(opts) do with {:ok, key} <- coerce_key(key) do Get.execute(cluster, key, :header, opts) end end @doc """ Reads multiple `keys` from `cluster` in one batch request per target node. The result list stays in the same order as `keys`. Each list item is either `{:ok, %Aerospike.Record{}}` for a hit or an indexed error for that key (`{:error, %Aerospike.Error{}}`, `{:error, :no_master}`, or `{:error, :unknown_node}`). Pass `:all` to read every bin, or pass a non-empty list of string or atom bin names to project only those bins. Batch opts include parent dispatch fields such as `:timeout`, `:socket_timeout`, `:max_concurrent_nodes`, `:allow_partial_results`, `:respond_all_keys`, `:allow_inline`, and `:allow_inline_ssd`, plus encodable read fields such as `:filter`, `:read_mode_ap`, `:read_mode_sc`, and `:read_touch_ttl_percent`. Accepts `%Aerospike.Key{}` values or `{namespace, set, user_key}` tuples. """ @spec batch_get(cluster(), [Key.key_input()], :all | [String.t() | atom()], batch_read_opts()) :: {:ok, [ {:ok, Aerospike.Record.t()} | {:error, Aerospike.Error.t()} | {:error, :no_master | :unknown_node} ]} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready} def batch_get(cluster, keys, bins \\ :all, opts \\ []) def batch_get(cluster, keys, bins, opts) when is_list(keys) do with {:ok, keys} <- coerce_keys(keys) do BatchGet.execute(cluster, keys, bins, opts) end end @doc """ Reads record headers for multiple `keys` from `cluster`. The result list stays in the same order as `keys`. Each list item is either `{:ok, %Aerospike.Record{bins: %{}}}` for a hit or an indexed error for that key (`{:error, %Aerospike.Error{}}`, `{:error, :no_master}`, or `{:error, :unknown_node}`). This helper accepts the same batch read opts as `batch_get/4`. Accepts `%Aerospike.Key{}` values or `{namespace, set, user_key}` tuples. """ @spec batch_get_header(cluster(), [Key.key_input()], batch_read_opts()) :: {:ok, [ {:ok, Aerospike.Record.t()} | {:error, Aerospike.Error.t()} | {:error, :no_master | :unknown_node} ]} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready} def batch_get_header(cluster, keys, opts \\ []) def batch_get_header(cluster, keys, opts) when is_list(keys) and is_list(opts) do with {:ok, keys} <- coerce_keys(keys) do BatchGet.execute(cluster, keys, :header, opts) end end @doc """ Checks existence for multiple `keys` from `cluster`. The result list stays in the same order as `keys`. Each list item is either `{:ok, true}` / `{:ok, false}` for the targeted key or an indexed error for that key (`{:error, %Aerospike.Error{}}`, `{:error, :no_master}`, or `{:error, :unknown_node}`). This helper accepts the same batch read opts as `batch_get/4`. Accepts `%Aerospike.Key{}` values or `{namespace, set, user_key}` tuples. """ @spec batch_exists(cluster(), [Key.key_input()], batch_read_opts()) :: {:ok, [ {:ok, boolean()} | {:error, Aerospike.Error.t()} | {:error, :no_master | :unknown_node} ]} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready} def batch_exists(cluster, keys, opts \\ []) def batch_exists(cluster, keys, opts) when is_list(keys) and is_list(opts) do with {:ok, keys} <- coerce_keys(keys) do BatchGet.execute(cluster, keys, :exists, opts) end end @doc """ Runs one read-only operation list for multiple `keys` from `cluster`. The result list stays in the same order as `keys`. Each list item is either `{:ok, %Aerospike.Record{}}` for a hit or an indexed error for that key (`{:error, %Aerospike.Error{}}`, `{:error, :no_master}`, or `{:error, :unknown_node}`). Missing keys remain explicit per-key errors; this helper does not collapse misses to `nil`. This helper accepts the same batch read opts as `batch_get/4`. Accepts `%Aerospike.Key{}` values or `{namespace, set, user_key}` tuples. """ @spec batch_get_operate(cluster(), [Key.key_input()], [Aerospike.Op.t()], batch_read_opts()) :: {:ok, [ {:ok, Aerospike.Record.t()} | {:error, Aerospike.Error.t()} | {:error, :no_master | :unknown_node} ]} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready} def batch_get_operate(cluster, keys, operations, opts \\ []) def batch_get_operate(cluster, keys, operations, opts) when is_list(keys) and is_list(operations) and is_list(opts) do with {:ok, keys} <- coerce_keys(keys) do BatchGet.execute_operate(cluster, keys, operations, opts) end end @doc """ Deletes multiple `keys` from `cluster`. The result list stays in the same order as `keys`. Each list item is a `%Aerospike.BatchResult{}` with `status: :ok` for a deleted record or `status: :error` for a per-key failure such as a missing key, routing failure, or node transport error. Missing keys remain explicit error results with the server result code; this helper does not collapse them to booleans. This helper accepts parent batch opts and encodable write fields such as `:generation_policy`, `:generation`, `:commit_level`, `:durable_delete`, `:send_key`, and `:filter`. Accepts `%Aerospike.Key{}` values or `{namespace, set, user_key}` tuples. """ @spec batch_delete(cluster(), [Key.key_input()], batch_write_opts()) :: {:ok, [Aerospike.BatchResult.t()]} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready} def batch_delete(cluster, keys, opts \\ []) def batch_delete(cluster, keys, opts) when is_list(keys) and is_list(opts) do with {:ok, keys} <- coerce_keys(keys) do BatchDelete.execute(cluster, keys, opts) end end @doc """ Executes one record UDF for multiple `keys` from `cluster`. The result list stays in the same order as `keys`. Each list item is a `%Aerospike.BatchResult{}` with `status: :ok` for a successful UDF row or `status: :error` for a per-key failure such as a missing key, missing UDF, routing failure, or node transport error. Successful UDF rows may include a returned `%Aerospike.Record{}` when the server sends return bins. This helper accepts the same parent batch and write opts as `batch_delete/3`. Accepts `%Aerospike.Key{}` values or `{namespace, set, user_key}` tuples. """ @spec batch_udf( cluster(), [Key.key_input()], String.t(), String.t(), list(), batch_write_opts() ) :: {:ok, [Aerospike.BatchResult.t()]} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready} def batch_udf(cluster, keys, package, function, args, opts \\ []) def batch_udf(cluster, keys, package, function, args, opts) when is_list(keys) and is_binary(package) and is_binary(function) and is_list(args) and is_list(opts) do with {:ok, keys} <- coerce_keys(keys) do BatchUdf.execute(cluster, keys, package, function, args, opts) end end @doc """ Executes heterogeneous batch entries built with `Aerospike.Batch`. The result list stays in the same order as the input entries. Each list item is a `%Aerospike.BatchResult{}` with `status: :ok` for a successful row or `status: :error` for a per-key failure such as a missing key, routing failure, server error, or node transport error. Parent batch policy opts are accepted in `opts`. Per-entry read/write policy opts can be attached with the `Aerospike.Batch` builders. """ @spec batch_operate(cluster(), [Batch.t()], batch_parent_opts()) :: {:ok, [Aerospike.BatchResult.t()]} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready} def batch_operate(cluster, entries, opts \\ []) def batch_operate(_cluster, [], opts) when is_list(opts) do with {:ok, _policy} <- Policy.batch(opts) do {:ok, []} end end def batch_operate(cluster, entries, opts) when is_list(entries) and is_list(opts) do with {:ok, _validated} <- Policy.batch(opts), {:ok, policy} <- batch_policy(cluster, opts), {:ok, command_entries} <- batch_command_entries(entries, policy), {:ok, results} <- MixedBatch.execute(cluster, command_entries, opts) do {:ok, MixedBatch.to_public_results(results)} end end @doc """ Returns a lazy `Stream` of records from a scan. The returned stream is lazy at the Enumerable boundary, but the current runtime drains each node response fully before yielding that node's records downstream. It does not promise frame-by-frame backpressure or an explicit cancellation API. Pass `node: node_name` in `opts` to scan one active node, using a name returned by `node_names/1` or `nodes/1`. """ @spec scan_stream(cluster(), Scan.t(), scan_query_opts()) :: {:ok, Enumerable.t()} | {:error, Aerospike.Error.t()} def scan_stream(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do ScanOps.stream(cluster, scan, opts) end @doc """ Same as `scan_stream/3` but raises on error. """ @spec scan_stream!(cluster(), Scan.t(), scan_query_opts()) :: Enumerable.t() def scan_stream!(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do case scan_stream(cluster, scan, opts) do {:ok, stream} -> stream {:error, %Aerospike.Error{} = err} -> raise err end end @doc """ Deprecated alias for `scan_stream!/3`. """ @deprecated "Use scan_stream!/3 instead." @spec stream!(cluster(), Scan.t(), scan_query_opts()) :: Enumerable.t() def stream!(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do scan_stream!(cluster, scan, opts) end @doc """ Returns a lazy `Stream` of records from a secondary-index query. Like `scan_stream/3`, this is lazy only at the outer Enumerable boundary. The current runtime buffers each node's query results before yielding them to the caller. Pass `node: node_name` in `opts` to query one active node, using a name returned by `node_names/1` or `nodes/1`. """ @spec query_stream(cluster(), Query.t(), scan_query_opts()) :: {:ok, Enumerable.t()} | {:error, Aerospike.Error.t()} def query_stream(cluster, %Query{} = query, opts \\ []) when is_list(opts) do ScanOps.query_stream(cluster, query, opts) end @doc """ Same as `query_stream/3` but raises on error. """ @spec query_stream!(cluster(), Query.t(), scan_query_opts()) :: Enumerable.t() def query_stream!(cluster, %Query{} = query, opts \\ []) when is_list(opts) do case query_stream(cluster, query, opts) do {:ok, stream} -> stream {:error, %Aerospike.Error{} = err} -> raise err end end @doc """ Eagerly collects scan records into a list. Pass `node: node_name` in `opts` to scan one active node, using a name returned by `node_names/1` or `nodes/1`. """ @spec scan_all(cluster(), Scan.t(), scan_query_opts()) :: {:ok, [Aerospike.Record.t()]} | {:error, Aerospike.Error.t()} def scan_all(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do ScanOps.all(cluster, scan, opts) end @doc """ Deprecated alias for `scan_all/3`. """ @deprecated "Use scan_all/3 instead." @spec all(cluster(), Scan.t(), scan_query_opts()) :: {:ok, [Aerospike.Record.t()]} | {:error, Aerospike.Error.t()} def all(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do scan_all(cluster, scan, opts) end @doc """ Eagerly collects query records into a list. `query.max_records` must be set because this helper advances through the query in repeated page-sized steps until the cursor is exhausted. Pass `node: node_name` in `opts` to query one active node, using a name returned by `node_names/1` or `nodes/1`. """ @spec query_all(cluster(), Query.t(), scan_query_opts()) :: {:ok, [Aerospike.Record.t()]} | {:error, Aerospike.Error.t()} def query_all(cluster, %Query{} = query, opts \\ []) when is_list(opts) do ScanOps.query_all(cluster, query, opts) end @doc """ Same as `query_all/3` but returns the list or raises `Aerospike.Error`. """ @spec query_all!(cluster(), Query.t(), scan_query_opts()) :: [Aerospike.Record.t()] def query_all!(cluster, %Query{} = query, opts \\ []) when is_list(opts) do case query_all(cluster, query, opts) do {:ok, records} -> records {:error, %Aerospike.Error{} = err} -> raise err end end @doc """ Same as `scan_all/3` but returns the list or raises `Aerospike.Error`. """ @spec scan_all!(cluster(), Scan.t(), scan_query_opts()) :: [Aerospike.Record.t()] def scan_all!(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do case scan_all(cluster, scan, opts) do {:ok, records} -> records {:error, %Aerospike.Error{} = err} -> raise err end end @doc """ Returns one collected scan page and a resumable cursor when more records remain. `scan.max_records` is required because it seeds the partition-tracker budget for the page walk. On multi-node scans that budget is distributed across active nodes, so a page is resumable but not guaranteed to contain exactly `scan.max_records` records. The cursor resumes partition progress from the prior page; it is not a stable snapshot token. Pass `node: node_name` in `opts` to collect a page from one active node, using a name returned by `node_names/1` or `nodes/1`. """ @spec scan_page(cluster(), Scan.t(), scan_query_opts()) :: {:ok, Page.t()} | {:error, Aerospike.Error.t()} def scan_page(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do ScanOps.scan_page(cluster, scan, opts) end @doc """ Same as `scan_page/3` but returns the page or raises `Aerospike.Error`. """ @spec scan_page!(cluster(), Scan.t(), scan_query_opts()) :: Page.t() def scan_page!(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do case scan_page(cluster, scan, opts) do {:ok, page} -> page {:error, %Aerospike.Error{} = err} -> raise err end end @doc """ Deprecated alias for `scan_all!/3`. """ @deprecated "Use scan_all!/3 instead." @spec all!(cluster(), Scan.t(), scan_query_opts()) :: [Aerospike.Record.t()] def all!(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do scan_all!(cluster, scan, opts) end @doc """ Counts scan matches without materializing the records. Pass `node: node_name` in `opts` to count records from one active node, using a name returned by `node_names/1` or `nodes/1`. """ @spec scan_count(cluster(), Scan.t(), scan_query_opts()) :: {:ok, non_neg_integer()} | {:error, Aerospike.Error.t()} def scan_count(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do ScanOps.count(cluster, scan, opts) end @doc """ Deprecated alias for `scan_count/3`. """ @deprecated "Use scan_count/3 instead." @spec count(cluster(), Scan.t(), scan_query_opts()) :: {:ok, non_neg_integer()} | {:error, Aerospike.Error.t()} def count(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do scan_count(cluster, scan, opts) end @doc """ Counts query matches without materializing the records. This still walks the query stream and counts client-side. It is not a separate server-side count primitive. Pass `node: node_name` in `opts` to count records from one active node, using a name returned by `node_names/1` or `nodes/1`. """ @spec query_count(cluster(), Query.t(), scan_query_opts()) :: {:ok, non_neg_integer()} | {:error, Aerospike.Error.t()} def query_count(cluster, %Query{} = query, opts \\ []) when is_list(opts) do ScanOps.query_count(cluster, query, opts) end @doc """ Same as `query_count/3` but returns the count or raises `Aerospike.Error`. """ @spec query_count!(cluster(), Query.t(), scan_query_opts()) :: non_neg_integer() def query_count!(cluster, %Query{} = query, opts \\ []) when is_list(opts) do case query_count(cluster, query, opts) do {:ok, count} -> count {:error, %Aerospike.Error{} = err} -> raise err end end @doc """ Sends one info command to one active cluster node and returns that node's reply. """ @spec info(cluster(), String.t(), admin_opts()) :: {:ok, String.t()} | {:error, Aerospike.Error.t()} def info(cluster, command, opts \\ []) when is_binary(command) and is_list(opts) do Admin.info(cluster, command, opts) end @doc """ Sends one info command to the named active cluster node and returns that node's reply. `node_name` must be one of the names returned by `node_names/1` or `nodes/1`. Stale or unknown names return an `%Aerospike.Error{code: :invalid_node}` instead of falling back to a different node. {:ok, [node_name | _]} = Aerospike.node_names(:aerospike) {:ok, response} = Aerospike.info_node(:aerospike, node_name, "statistics") Supported options: * `:pool_checkout_timeout` — non-negative pool checkout timeout in milliseconds. """ @spec info_node(cluster(), String.t(), String.t(), admin_opts()) :: {:ok, String.t()} | {:error, Aerospike.Error.t()} def info_node(cluster, node_name, command, opts \\ []) when is_binary(node_name) and is_binary(command) and is_list(opts) do Admin.info_node(cluster, node_name, command, opts) end @doc """ Returns the published active cluster nodes with their direct-connect host and port. """ @spec nodes(cluster()) :: {:ok, [node_info()]} def nodes(cluster) do {:ok, Cluster.nodes(cluster)} end @doc """ Returns the published active cluster node-name snapshot. """ @spec node_names(cluster()) :: {:ok, [String.t()]} def node_names(cluster) do {:ok, Cluster.node_names(cluster)} end @doc """ Returns whether internal runtime metrics are enabled for `cluster`. Metrics are opt-in. The collector is initialized at cluster start so the config rows exist early, but command, pool, and tender counters remain idle until `enable_metrics/2` is called. """ @spec metrics_enabled?(cluster()) :: boolean() def metrics_enabled?(cluster) do RuntimeMetrics.metrics_enabled?(cluster) end @doc """ Enables internal runtime metrics for `cluster`. Supported options: * `:reset` — boolean. When `true`, clears the existing runtime counters before enabling collection. """ @spec enable_metrics(cluster(), metrics_opts()) :: :ok | {:error, Aerospike.Error.t()} def enable_metrics(cluster, opts \\ []) when is_list(opts) do with :ok <- validate_enable_metrics_opts(opts) do RuntimeMetrics.enable(cluster, opts) end end @doc """ Disables internal runtime metrics for `cluster`. Counter rows remain in place so `stats/1` still reports the last collected values until metrics are re-enabled or reset. """ @spec disable_metrics(cluster()) :: :ok | {:error, Aerospike.Error.t()} def disable_metrics(cluster) do RuntimeMetrics.disable(cluster) end @doc """ Returns the current internal runtime metrics snapshot for `cluster`. The returned map is intentionally limited to the counters and cluster metadata the runtime collector actually records today. It is not a generalized exporter surface. """ @spec stats(cluster()) :: map() def stats(cluster) do RuntimeMetrics.stats(cluster) end @doc """ Verifies that the active node pools can serve checkouts through the normal path. This is an explicit operator action; it does not toggle metrics and it does not change the pool startup mode. The current driver pools are already eager at cluster start. `warm_up/2` simply proves that the active pools can hand out the requested number of workers right now. Supported options: * `:count` — non-negative integer. `0` (default) means "up to the configured pool size per active node". * `:pool_checkout_timeout` — non-negative integer timeout in milliseconds for each checkout probe. """ @spec warm_up(cluster(), warm_up_opts()) :: {:ok, map()} | {:error, Aerospike.Error.t()} def warm_up(cluster, opts \\ []) when is_list(opts) do with :ok <- validate_warm_up_opts(opts) do Cluster.warm_up(cluster, opts) end end @doc """ Creates a secondary index and returns a pollable task handle. Required options: * `:bin` — non-empty bin name. * `:name` — non-empty index name. * `:type` — one of `:numeric`, `:string`, or `:geo2dsphere`. Optional options: * `:collection` — one of `:list`, `:mapkeys`, or `:mapvalues`. * `:ctx` — non-empty nested CDT path built with `Aerospike.Ctx`. * `:pool_checkout_timeout` — non-negative pool checkout timeout in milliseconds. Geo indexes use `type: :geo2dsphere` and can be queried with `Aerospike.Filter.geo_within/2` or `Aerospike.Filter.geo_contains/2`. {:ok, task} = Aerospike.create_index(cluster, "test", "places", bin: "loc", name: "places_loc_geo_idx", type: :geo2dsphere ) :ok = Aerospike.IndexTask.wait(task) """ @spec create_index(cluster(), String.t(), String.t(), create_index_opts()) :: {:ok, IndexTask.t()} | {:error, Aerospike.Error.t()} def create_index(cluster, namespace, set, opts \\ []) when is_binary(namespace) and is_binary(set) and is_list(opts) do Admin.create_index(cluster, namespace, set, opts) end @doc """ Creates an expression-backed secondary index and returns a pollable task handle. Required options: * `:name` — non-empty index name. * `:type` — one of `:numeric`, `:string`, or `:geo2dsphere`. Optional options: * `:collection` — one of `:list`, `:mapkeys`, or `:mapvalues`. * `:pool_checkout_timeout` — non-negative pool checkout timeout in milliseconds. The source must be a `%Aerospike.Exp{}` with non-empty wire bytes. Expression indexes use the expression as the source and therefore do not accept `:bin`. Servers older than Aerospike 8.1 reject expression-backed index creation before the create command is sent. {:ok, task} = Aerospike.create_expression_index(cluster, "test", "users", Exp.int_bin("age"), name: "users_age_expr_idx", type: :numeric ) :ok = Aerospike.IndexTask.wait(task) """ @spec create_expression_index( cluster(), String.t(), String.t(), Exp.t(), create_expression_index_opts() ) :: {:ok, IndexTask.t()} | {:error, Aerospike.Error.t()} def create_expression_index(cluster, namespace, set, %Exp{} = expression, opts \\ []) when is_binary(namespace) and is_binary(set) and is_list(opts) do with {:ok, _policy} <- Policy.expression_index_create(expression, opts) do Admin.create_expression_index(cluster, namespace, set, expression, opts) end end @doc """ Sets or clears the Enterprise XDR filter for one datacenter and namespace. Pass a non-empty `%Aerospike.Exp{}` to set the filter, or `nil` to clear the current filter. `datacenter` and `namespace` must be non-empty info-command identifiers and cannot contain command delimiters. Live application requires an Enterprise server with XDR configured. Community Edition or unconfigured clusters may reject the command after local validation. filter = Exp.eq(Exp.int_bin("active"), Exp.int(1)) :ok = Aerospike.set_xdr_filter(cluster, "dc-west", "test", filter) :ok = Aerospike.set_xdr_filter(cluster, "dc-west", "test", nil) """ @spec set_xdr_filter(cluster(), String.t(), String.t(), Exp.t() | nil) :: :ok | {:error, Aerospike.Error.t()} def set_xdr_filter(cluster, datacenter, namespace, filter) when is_binary(datacenter) and is_binary(namespace) do with :ok <- Policy.xdr_filter(datacenter, namespace, filter) do Admin.set_xdr_filter(cluster, datacenter, namespace, filter) end end @doc """ Drops a secondary index. """ @spec drop_index(cluster(), String.t(), String.t(), admin_opts()) :: :ok | {:error, Aerospike.Error.t()} def drop_index(cluster, namespace, index_name, opts \\ []) when is_binary(namespace) and is_binary(index_name) and is_list(opts) do Admin.drop_index(cluster, namespace, index_name, opts) end @doc """ Lists the registered server-side UDF packages visible from one active node. This is package lifecycle state, not record execution. Use `apply_udf/6` to invoke one function against one key. """ @spec list_udfs(cluster(), admin_opts()) :: {:ok, [UDF.t()]} | {:error, Aerospike.Error.t()} def list_udfs(cluster, opts \\ []) when is_list(opts) do Admin.list_udfs(cluster, opts) end @doc """ Uploads a UDF package from inline source or a readable local `.lua` path. Returns a pollable `Aerospike.RegisterTask` once the server accepts the upload. The package may still be propagating, so call `RegisterTask.wait/2` before relying on it from `apply_udf/6` or background UDF jobs. """ @spec register_udf(cluster(), String.t(), String.t(), admin_opts()) :: {:ok, RegisterTask.t()} | {:error, Aerospike.Error.t()} def register_udf(cluster, path_or_content, server_name, opts \\ []) when is_binary(path_or_content) and is_binary(server_name) and is_list(opts) do Admin.register_udf(cluster, path_or_content, server_name, opts) end @doc """ Removes a registered UDF package by server filename. This is idempotent: removing an already-absent package still returns `:ok`. """ @spec remove_udf(cluster(), String.t(), admin_opts()) :: :ok | {:error, Aerospike.Error.t()} def remove_udf(cluster, server_name, opts \\ []) when is_binary(server_name) and is_list(opts) do Admin.remove_udf(cluster, server_name, opts) end @doc """ Returns one collected query page and a resumable cursor when more records remain. `query.max_records` is required because it seeds the partition-tracker budget for the page walk. On multi-node queries that budget is distributed across active nodes, so a page is resumable but not guaranteed to contain exactly `query.max_records` records. The cursor resumes partition progress from the prior page; it is not a stable snapshot token. Pass `node: node_name` in `opts` to collect a page from one active node, using a name returned by `node_names/1` or `nodes/1`. """ @spec query_page(cluster(), Query.t(), scan_query_opts()) :: {:ok, Page.t()} | {:error, Aerospike.Error.t()} def query_page(cluster, %Query{} = query, opts \\ []) when is_list(opts) do ScanOps.query_page(cluster, query, opts) end @doc """ Same as `query_page/3` but returns the page or raises `Aerospike.Error`. """ @spec query_page!(cluster(), Query.t(), scan_query_opts()) :: Page.t() def query_page!(cluster, %Query{} = query, opts \\ []) when is_list(opts) do case query_page(cluster, query, opts) do {:ok, page} -> page {:error, %Aerospike.Error{} = err} -> raise err end end @doc """ Truncates all records in `namespace`. This sends one truncate info command through the shared admin seam. The server then distributes the truncate across the cluster. Options: * `:before` — `%DateTime{}` — truncate only records whose last-update time is older than the provided timestamp * `:pool_checkout_timeout` — non-negative integer checkout timeout in milliseconds for the one-node admin request """ @spec truncate(cluster(), String.t(), truncate_opts()) :: :ok | {:error, Aerospike.Error.t()} def truncate(cluster, namespace, opts \\ []) def truncate(cluster, namespace, set) when is_binary(namespace) and is_binary(set) do truncate(cluster, namespace, set, []) end def truncate(cluster, namespace, opts) when is_binary(namespace) and is_list(opts) do with {:ok, opts} <- validate_truncate_opts(opts, "Aerospike.truncate/3") do Admin.truncate(cluster, namespace, opts) end end @doc """ Truncates all records in `namespace` and `set`. Like `truncate/3`, this uses the shared one-node admin info seam. The optional `:before` filter is forwarded to the server as a last-update cutoff. Options: * `:before` — `%DateTime{}` — truncate only records whose last-update time is older than the provided timestamp * `:pool_checkout_timeout` — non-negative integer checkout timeout in milliseconds for the one-node admin request """ @spec truncate(cluster(), String.t(), String.t(), truncate_opts()) :: :ok | {:error, Aerospike.Error.t()} def truncate(cluster, namespace, set, opts) when is_binary(namespace) and is_binary(set) and is_list(opts) do with {:ok, opts} <- validate_truncate_opts(opts, "Aerospike.truncate/4") do Admin.truncate(cluster, namespace, set, opts) end end @doc """ Creates a password-authenticated security user. This requires Aerospike Enterprise with security enabled and a cluster connection authenticated as a user that holds the `user-admin` privilege. Supported opts are: * `:timeout` * `:pool_checkout_timeout` """ @spec create_user(cluster(), String.t(), String.t(), [String.t()], security_admin_opts()) :: :ok | {:error, Aerospike.Error.t()} def create_user(cluster, user_name, password, roles, opts \\ []) when is_binary(user_name) and is_binary(password) and is_list(roles) and is_list(opts) do with {:ok, role_names} <- validate_role_names(roles), {:ok, _policy} <- Policy.security_admin(opts) do Admin.create_user(cluster, user_name, password, role_names, opts) end end @doc """ Creates a PKI-authenticated security user. The user is created with a no-password credential and is intended for TLS certificate authentication. This requires Aerospike Enterprise with security enabled, server support for PKI users, and a cluster connection authenticated as a user that holds the `user-admin` privilege. Supported opts are: * `:timeout` * `:pool_checkout_timeout` """ @spec create_pki_user(cluster(), String.t(), [String.t()], security_admin_opts()) :: :ok | {:error, Aerospike.Error.t()} def create_pki_user(cluster, user_name, roles, opts \\ []) when is_binary(user_name) and is_list(roles) and is_list(opts) do with {:ok, role_names} <- validate_role_names(roles), {:ok, _policy} <- Policy.security_admin(opts) do Admin.create_pki_user(cluster, user_name, role_names, opts) end end @doc """ Drops a security user. This requires Aerospike Enterprise with security enabled. """ @spec drop_user(cluster(), String.t(), security_admin_opts()) :: :ok | {:error, Aerospike.Error.t()} def drop_user(cluster, user_name, opts \\ []) when is_binary(user_name) and is_list(opts) do with {:ok, _policy} <- Policy.security_admin(opts) do Admin.drop_user(cluster, user_name, opts) end end @doc """ Changes a security user's password. When `user_name` matches the credentials configured on `cluster`, the driver uses the self-service password-change command and rotates the running cluster's in-memory credentials for future reconnects. For other users it uses the user-admin password-set command. This requires Aerospike Enterprise with security enabled. """ @spec change_password(cluster(), String.t(), String.t(), security_admin_opts()) :: :ok | {:error, Aerospike.Error.t()} def change_password(cluster, user_name, password, opts \\ []) when is_binary(user_name) and is_binary(password) and is_list(opts) do with {:ok, _policy} <- Policy.security_admin(opts) do Admin.change_password(cluster, user_name, password, opts) end end @doc """ Grants roles to a security user. This requires Aerospike Enterprise with security enabled. """ @spec grant_roles(cluster(), String.t(), [String.t()], security_admin_opts()) :: :ok | {:error, Aerospike.Error.t()} def grant_roles(cluster, user_name, roles, opts \\ []) when is_binary(user_name) and is_list(roles) and is_list(opts) do with {:ok, role_names} <- validate_role_names(roles), {:ok, _policy} <- Policy.security_admin(opts) do Admin.grant_roles(cluster, user_name, role_names, opts) end end @doc """ Revokes roles from a security user. This requires Aerospike Enterprise with security enabled. """ @spec revoke_roles(cluster(), String.t(), [String.t()], security_admin_opts()) :: :ok | {:error, Aerospike.Error.t()} def revoke_roles(cluster, user_name, roles, opts \\ []) when is_binary(user_name) and is_list(roles) and is_list(opts) do with {:ok, role_names} <- validate_role_names(roles), {:ok, _policy} <- Policy.security_admin(opts) do Admin.revoke_roles(cluster, user_name, role_names, opts) end end @doc """ Queries one security user. Returns `{:ok, nil}` when the named user does not exist. This requires Aerospike Enterprise with security enabled. """ @spec query_user(cluster(), String.t(), security_admin_opts()) :: {:ok, user_info() | nil} | {:error, Aerospike.Error.t()} def query_user(cluster, user_name, opts \\ []) when is_binary(user_name) and is_list(opts) do with {:ok, _policy} <- Policy.security_admin(opts) do Admin.query_user(cluster, user_name, opts) end end @doc """ Queries all security users visible to the authenticated cluster user. This requires Aerospike Enterprise with security enabled. """ @spec query_users(cluster(), security_admin_opts()) :: {:ok, [user_info()]} | {:error, Aerospike.Error.t()} def query_users(cluster, opts \\ []) when is_list(opts) do with {:ok, _policy} <- Policy.security_admin(opts) do Admin.query_users(cluster, opts) end end @doc """ Creates a security role. This requires Aerospike Enterprise with security enabled. Supported opts are: * `:whitelist` — list of client address strings * `:read_quota` — non-negative integer operations-per-second limit * `:write_quota` — non-negative integer operations-per-second limit * `:timeout` * `:pool_checkout_timeout` """ @spec create_role(cluster(), String.t(), [privilege()], create_role_opts()) :: :ok | {:error, Aerospike.Error.t()} def create_role(cluster, role_name, privileges, opts \\ []) when is_binary(role_name) and is_list(privileges) and is_list(opts) do with {:ok, privileges} <- validate_privileges(privileges), {:ok, {whitelist, read_quota, write_quota, call_opts}} <- validate_create_role_opts(opts), {:ok, _policy} <- Policy.security_admin(call_opts) do Admin.create_role( cluster, role_name, privileges, whitelist, read_quota, write_quota, call_opts ) end end @doc """ Drops a security role. This requires Aerospike Enterprise with security enabled. """ @spec drop_role(cluster(), String.t(), security_admin_opts()) :: :ok | {:error, Aerospike.Error.t()} def drop_role(cluster, role_name, opts \\ []) when is_binary(role_name) and is_list(opts) do with {:ok, _policy} <- Policy.security_admin(opts) do Admin.drop_role(cluster, role_name, opts) end end @doc """ Sets or clears a security role's client-address whitelist. Pass an empty list to clear the role's whitelist. This requires Aerospike Enterprise with security enabled. """ @spec set_whitelist(cluster(), String.t(), [String.t()], security_admin_opts()) :: :ok | {:error, Aerospike.Error.t()} def set_whitelist(cluster, role_name, whitelist, opts \\ []) when is_binary(role_name) and is_list(opts) do with {:ok, whitelist} <- validate_role_whitelist_value(whitelist), {:ok, _policy} <- Policy.security_admin(opts) do Admin.set_whitelist(cluster, role_name, whitelist, opts) end end @doc """ Sets read and write quota limits for a security role. Pass `0` for either quota to clear that limit. Quotas require server security configuration with quotas enabled. """ @spec set_quotas( cluster(), String.t(), non_neg_integer(), non_neg_integer(), security_admin_opts() ) :: :ok | {:error, Aerospike.Error.t()} def set_quotas(cluster, role_name, read_quota, write_quota, opts \\ []) when is_binary(role_name) and is_list(opts) do with {:ok, read_quota} <- validate_role_quota_value(read_quota, :read_quota), {:ok, write_quota} <- validate_role_quota_value(write_quota, :write_quota), {:ok, _policy} <- Policy.security_admin(opts) do Admin.set_quotas(cluster, role_name, read_quota, write_quota, opts) end end @doc """ Grants privileges to a security role. This requires Aerospike Enterprise with security enabled. """ @spec grant_privileges(cluster(), String.t(), [privilege()], security_admin_opts()) :: :ok | {:error, Aerospike.Error.t()} def grant_privileges(cluster, role_name, privileges, opts \\ []) when is_binary(role_name) and is_list(privileges) and is_list(opts) do with {:ok, privileges} <- validate_privileges(privileges), {:ok, _policy} <- Policy.security_admin(opts) do Admin.grant_privileges(cluster, role_name, privileges, opts) end end @doc """ Revokes privileges from a security role. This requires Aerospike Enterprise with security enabled. """ @spec revoke_privileges(cluster(), String.t(), [privilege()], security_admin_opts()) :: :ok | {:error, Aerospike.Error.t()} def revoke_privileges(cluster, role_name, privileges, opts \\ []) when is_binary(role_name) and is_list(privileges) and is_list(opts) do with {:ok, privileges} <- validate_privileges(privileges), {:ok, _policy} <- Policy.security_admin(opts) do Admin.revoke_privileges(cluster, role_name, privileges, opts) end end @doc """ Queries one security role. Returns `{:ok, nil}` when the named role does not exist. This requires Aerospike Enterprise with security enabled. """ @spec query_role(cluster(), String.t(), security_admin_opts()) :: {:ok, role_info() | nil} | {:error, Aerospike.Error.t()} def query_role(cluster, role_name, opts \\ []) when is_binary(role_name) and is_list(opts) do with {:ok, _policy} <- Policy.security_admin(opts) do Admin.query_role(cluster, role_name, opts) end end @doc """ Queries all security roles visible to the authenticated cluster user. This requires Aerospike Enterprise with security enabled. """ @spec query_roles(cluster(), security_admin_opts()) :: {:ok, [role_info()]} | {:error, Aerospike.Error.t()} def query_roles(cluster, opts \\ []) when is_list(opts) do with {:ok, _policy} <- Policy.security_admin(opts) do Admin.query_roles(cluster, opts) end end @doc """ Streams aggregate query values over the same node-buffered query runtime used by `query_stream/3`. The returned stream yields the aggregate values emitted by the server. It does not run local Lua finalization. Use `query_aggregate_result/6` when the caller wants one locally finalized aggregate result. """ @spec query_aggregate(cluster(), Query.t(), String.t(), String.t(), list(), scan_query_opts()) :: {:ok, Enumerable.t()} | {:error, Aerospike.Error.t()} def query_aggregate(cluster, %Query{} = query, package, function, args, opts \\ []) when is_binary(package) and is_binary(function) and is_list(args) and is_list(opts) do ScanOps.query_aggregate(cluster, query, package, function, args, opts) end @doc """ Returns one finalized aggregate query result. This runs the same server aggregate query as `query_aggregate/6`, then locally executes the package's Lua stream finalization over the server-emitted aggregate values. Use `query_aggregate/6` when callers need the partial server values themselves. The local Lua package source must be supplied with exactly one of: * `:source` - inline Lua source as a binary * `:source_path` - readable local Lua source path Source loading is local-only and separate from server UDF registration. The client does not derive a local path from `package` and does not fetch source from the server. Source option errors, unsupported local argument values, and `node: node_name` fail with `{:error, %Aerospike.Error{code: :invalid_argument}}` before opening the server query stream. The local reducer runs in a fresh bounded Lua state. The existing `:timeout` option bounds local execution when it is a positive integer; otherwise a finite default is used. Supported local stream helpers are `map`, `filter`, `aggregate`, and `reduce`. Logging helpers are no-ops. Filesystem, OS, package loading, dynamic loading, debug access, `require`, `groupby`, `list`, `bytes`, and record/database mutation or lookup helpers fail explicitly with `%Aerospike.Error{code: :query_generic}`. Values crossing the local Lua boundary are limited to `nil`, booleans, integers, floats, binaries, lists, and maps with scalar keys. Blob, geo, raw, HLL, and other unsupported values fail instead of being coerced. Example: query = Aerospike.Query.new("test", "users") |> Aerospike.Query.where(Aerospike.Filter.range("age", 18, 65)) Aerospike.query_aggregate_result( :aerospike, query, "user_stats", "sum_age", ["age"], source_path: "priv/udf/user_stats.lua", timeout: 10_000 ) Returns `{:ok, nil}` when local finalization produces no value. Returns an error when finalization produces multiple values or local Lua execution fails. """ @spec query_aggregate_result( cluster(), Query.t(), String.t(), String.t(), list(), aggregate_result_opts() ) :: {:ok, term() | nil} | {:error, Aerospike.Error.t()} def query_aggregate_result(cluster, %Query{} = query, package, function, args, opts \\ []) when is_binary(package) and is_binary(function) and is_list(args) and is_list(opts) do ScanOps.query_aggregate_result(cluster, query, package, function, args, opts) end @doc """ Same as `query_aggregate_result/6` but returns the value or raises `Aerospike.Error`. """ @spec query_aggregate_result!( cluster(), Query.t(), String.t(), String.t(), list(), aggregate_result_opts() ) :: term() | nil def query_aggregate_result!(cluster, %Query{} = query, package, function, args, opts \\ []) when is_binary(package) and is_binary(function) and is_list(args) and is_list(opts) do case query_aggregate_result(cluster, query, package, function, args, opts) do {:ok, result} -> result {:error, %Aerospike.Error{} = err} -> raise err end end @doc """ Starts a background query write job that applies the given operations. This returns a pollable task handle, not a resumable record stream. Pass `node: node_name` in `opts` to start the job on one active node, using a name returned by `node_names/1` or `nodes/1`. """ @spec query_execute(cluster(), Query.t(), list(), scan_query_opts()) :: {:ok, ExecuteTask.t()} | {:error, Aerospike.Error.t()} def query_execute(cluster, %Query{} = query, ops, opts \\ []) when is_list(ops) and is_list(opts) do ScanOps.query_execute(cluster, query, ops, opts) end @doc """ Starts a background query UDF job. This returns a pollable task handle, not a resumable record stream. Pass `node: node_name` in `opts` to start the job on one active node, using a name returned by `node_names/1` or `nodes/1`. """ @spec query_udf(cluster(), Query.t(), String.t(), String.t(), list(), scan_query_opts()) :: {:ok, ExecuteTask.t()} | {:error, Aerospike.Error.t()} def query_udf(cluster, %Query{} = query, package, function, args, opts \\ []) when is_binary(package) and is_binary(function) and is_list(args) and is_list(opts) do ScanOps.query_udf(cluster, query, package, function, args, opts) end @doc """ Executes one record UDF against `key`. This is a single-record command, not a background query job. It accepts the same write opts as the unary write family: * `:timeout` * `:socket_timeout` * `:max_retries` * `:sleep_between_retries_ms` * `:ttl` * `:generation` * `:generation_policy` * `:exists` * `:commit_level` * `:durable_delete` * `:respond_per_op` * `:send_key` * `:read_mode_ap` * `:read_mode_sc` * `:read_touch_ttl_percent` * `:use_compression` * `:filter` * `:txn` Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`. Transport failures are not retried automatically once the request is on the wire, because record UDFs may already have produced server-side effects. Package lifecycle lives on `register_udf/*`, `remove_udf/*`, and `list_udfs/2`. """ @spec apply_udf(cluster(), Key.key_input(), String.t(), String.t(), list(), write_opts()) :: {:ok, term()} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready | :no_master | :unknown_node} def apply_udf(cluster, key, package, function, args, opts \\ []) when is_binary(package) and is_binary(function) and is_list(args) and is_list(opts) do with {:ok, key} <- coerce_key(key) do ApplyUdf.execute(cluster, key, package, function, args, opts) end end @doc """ Commits a transaction on the named cluster `cluster`. This only works for a transaction handle whose tracking row is already initialized on `cluster`. A fresh `%Aerospike.Txn{}` is not enough by itself. In the current driver, public code initializes that runtime state only when `transaction/2` or `transaction/3` enters its callback. Transaction tracking is keyed off the started cluster name, so this helper currently requires that registered atom. """ @spec commit(named_cluster(), Txn.t()) :: {:ok, :committed | :already_committed} | {:error, Aerospike.Error.t()} def commit(cluster, %Txn{} = txn) when is_atom(cluster) do TxnRoll.commit(cluster, txn, []) end @doc """ Aborts a transaction on the named cluster `cluster`. Like `commit/2`, this requires a handle with initialized runtime tracking on `cluster`. It is for an already-open transaction; it does not create one, and it currently requires the registered cluster atom. """ @spec abort(named_cluster(), Txn.t()) :: {:ok, :aborted | :already_aborted} | {:error, Aerospike.Error.t()} def abort(cluster, %Txn{} = txn) when is_atom(cluster) do TxnRoll.abort(cluster, txn, []) end @doc """ Returns the current state of a transaction on the named cluster `cluster`. This reflects only the in-flight states backed by the runtime tracking row. After commit or abort, the driver cleans that row up, so `txn_status/2` returns an error instead of a terminal `:committed` or `:aborted` state. Like the other transaction lifecycle helpers, this currently requires the registered cluster atom. """ @spec txn_status(named_cluster(), Txn.t()) :: {:ok, :open | :verified | :committed | :aborted} | {:error, Aerospike.Error.t()} def txn_status(cluster, %Txn{} = txn) when is_atom(cluster) do TxnRoll.txn_status(cluster, txn) end @doc """ Runs a function within a new transaction on the named cluster `cluster`. The callback owns the public transaction lifecycle. The driver initializes the runtime tracking row before invoking `fun`, then commits on success or aborts on any failure path. Do not call `commit/2` or `abort/2` from inside the callback. The `%Aerospike.Txn{}` passed to `fun` is safe only for sequential use within that transaction. Do not share it across concurrent processes, and do not use scans or queries with it; the current transaction proof covers only transaction-aware single-record commands. This helper currently requires the registered cluster atom. """ @spec transaction(named_cluster(), (Txn.t() -> term())) :: {:ok, term()} | {:error, Aerospike.Error.t()} def transaction(cluster, fun) when is_atom(cluster) and is_function(fun, 1) do TxnRoll.transaction(cluster, [], fun) end @doc """ Runs a function within a transaction on the named cluster `cluster` using a provided handle or options. When `txn_or_opts` is a `%Aerospike.Txn{}`, the driver initializes fresh runtime tracking for that handle on `cluster` at callback entry. Reusing the same handle concurrently or against another cluster is unsupported. This helper currently requires the registered cluster atom. """ @spec transaction(named_cluster(), Txn.t() | transaction_opts(), (Txn.t() -> term())) :: {:ok, term()} | {:error, Aerospike.Error.t()} def transaction(cluster, txn_or_opts, fun) when is_atom(cluster) and is_function(fun, 1) do TxnRoll.transaction(cluster, txn_or_opts, fun) end @doc """ Same as `scan_count/3` but returns the count or raises `Aerospike.Error`. """ @spec scan_count!(cluster(), Scan.t(), scan_query_opts()) :: non_neg_integer() def scan_count!(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do case scan_count(cluster, scan, opts) do {:ok, count} -> count {:error, %Aerospike.Error{} = err} -> raise err end end @doc """ Deprecated alias for `scan_count!/3`. """ @deprecated "Use scan_count!/3 instead." @spec count!(cluster(), Scan.t(), scan_query_opts()) :: non_neg_integer() def count!(cluster, %Scan{} = scan, opts \\ []) when is_list(opts) do scan_count!(cluster, scan, opts) end @doc """ Writes `bins` for `key` to `cluster`. The driver accepts only a non-empty bin map and only scalar, list, map, bytes, geo, and HyperLogLog values supported by the command encoder. Supported write opts include: * `:timeout` * `:socket_timeout` * `:max_retries` * `:sleep_between_retries_ms` * `:ttl` * `:generation` * `:generation_policy` — `:none`, `:expect_equal`, or `:expect_gt` * `:exists` — one of `:update`, `:update_only`, `:create_or_replace`, `:replace_only`, or `:create_only` * `:commit_level` — `:all` or `:master` * `:durable_delete` — when `true`, write/delete commands that remove a record ask the server to leave a tombstone * `:respond_per_op` * `:send_key` * `:read_mode_ap` * `:read_mode_sc` * `:read_touch_ttl_percent` * `:use_compression` * `:filter` — non-empty `%Aerospike.Exp{}` server-side filter expression, or `nil` for no filter Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`. """ @spec put(cluster(), Key.key_input(), Aerospike.Record.bins_input(), write_opts()) :: {:ok, Aerospike.Record.metadata()} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready | :no_master | :unknown_node} def put(cluster, key, bins, opts \\ []) do with {:ok, key} <- coerce_key(key) do Put.execute(cluster, key, bins, opts) end end @doc """ Sends a caller-built single-record write/delete frame for `key`. This helper is intended for tooling, proxy, and replay scenarios. `payload` must be a complete Aerospike wire frame for one write-shaped command. The client uses `key` only to choose the write partition owner, forwards `payload` unchanged, and parses only the standard write response. The payload must already contain every server-visible write attribute, such as generation, TTL, send-key, delete flags, filters, and any transaction fields. Passing `:txn` validates the transaction option shape but does not register the key with the transaction monitor or add transaction fields. Supported write opts are validated for routing and I/O budgets: * `:timeout` * `:socket_timeout` * `:max_retries` * `:sleep_between_retries_ms` * `:ttl` * `:generation` * `:generation_policy` * `:exists` * `:commit_level` * `:durable_delete` * `:respond_per_op` * `:send_key` * `:use_compression` * `:filter` * `:txn` Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`. """ @spec put_payload(cluster(), Key.key_input(), binary(), write_opts()) :: :ok | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready | :no_master | :unknown_node} def put_payload(cluster, key, payload, opts \\ []) when is_binary(payload) and is_list(opts) do with {:ok, key} <- coerce_key(key) do PutPayload.execute(cluster, key, payload, opts) end end @doc """ Same as `put_payload/4` but returns `:ok` or raises `Aerospike.Error`. """ @spec put_payload!(cluster(), Key.key_input(), binary(), write_opts()) :: :ok def put_payload!(cluster, key, payload, opts \\ []) when is_binary(payload) and is_list(opts) do case put_payload(cluster, key, payload, opts) do :ok -> :ok {:error, %Aerospike.Error{} = err} -> raise err end end @doc """ Atomically adds numeric deltas in `bins` for `key`. This is a thin unary write helper over the same routed write path as `put/4`. The return shape stays aligned with the write family and does not expose `operate/4` record results. Supported write opts include: * `:timeout` * `:socket_timeout` * `:max_retries` * `:sleep_between_retries_ms` * `:ttl` * `:generation` * `:generation_policy` — `:none`, `:expect_equal`, or `:expect_gt` * `:exists` — one of `:update`, `:update_only`, `:create_or_replace`, `:replace_only`, or `:create_only` * `:commit_level` — `:all` or `:master` * `:durable_delete` — when `true`, write/delete commands that remove a record ask the server to leave a tombstone * `:respond_per_op` * `:send_key` * `:read_mode_ap` * `:read_mode_sc` * `:read_touch_ttl_percent` * `:use_compression` * `:filter` — non-empty `%Aerospike.Exp{}` server-side filter expression, or `nil` for no filter Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`. """ @spec add(cluster(), Key.key_input(), Aerospike.Record.bins_input(), write_opts()) :: {:ok, Aerospike.Record.metadata()} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready | :no_master | :unknown_node} def add(cluster, key, bins, opts \\ []) do with {:ok, key} <- coerce_key(key) do WriteOp.execute(cluster, key, :add, bins, opts) end end @doc """ Atomically appends string suffixes in `bins` for `key`. This stays on the unary write path and returns write metadata, not an `operate/4` record payload. Supported write opts include: * `:timeout` * `:socket_timeout` * `:max_retries` * `:sleep_between_retries_ms` * `:ttl` * `:generation` * `:generation_policy` — `:none`, `:expect_equal`, or `:expect_gt` * `:exists` — one of `:update`, `:update_only`, `:create_or_replace`, `:replace_only`, or `:create_only` * `:commit_level` — `:all` or `:master` * `:durable_delete` — when `true`, write/delete commands that remove a record ask the server to leave a tombstone * `:respond_per_op` * `:send_key` * `:read_mode_ap` * `:read_mode_sc` * `:read_touch_ttl_percent` * `:use_compression` * `:filter` — non-empty `%Aerospike.Exp{}` server-side filter expression, or `nil` for no filter Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`. """ @spec append(cluster(), Key.key_input(), Aerospike.Record.bins_input(), write_opts()) :: {:ok, Aerospike.Record.metadata()} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready | :no_master | :unknown_node} def append(cluster, key, bins, opts \\ []) do with {:ok, key} <- coerce_key(key) do WriteOp.execute(cluster, key, :append, bins, opts) end end @doc """ Atomically prepends string prefixes in `bins` for `key`. This stays on the unary write path and returns write metadata, not an `operate/4` record payload. Supported write opts include: * `:timeout` * `:socket_timeout` * `:max_retries` * `:sleep_between_retries_ms` * `:ttl` * `:generation` * `:generation_policy` — `:none`, `:expect_equal`, or `:expect_gt` * `:exists` — one of `:update`, `:update_only`, `:create_or_replace`, `:replace_only`, or `:create_only` * `:commit_level` — `:all` or `:master` * `:durable_delete` — when `true`, write/delete commands that remove a record ask the server to leave a tombstone * `:respond_per_op` * `:send_key` * `:read_mode_ap` * `:read_mode_sc` * `:read_touch_ttl_percent` * `:use_compression` * `:filter` — non-empty `%Aerospike.Exp{}` server-side filter expression, or `nil` for no filter Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`. """ @spec prepend(cluster(), Key.key_input(), Aerospike.Record.bins_input(), write_opts()) :: {:ok, Aerospike.Record.metadata()} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready | :no_master | :unknown_node} def prepend(cluster, key, bins, opts \\ []) do with {:ok, key} <- coerce_key(key) do WriteOp.execute(cluster, key, :prepend, bins, opts) end end @doc """ Returns whether `key` exists in `cluster` without reading bins. Supported read opts are `:timeout`, `:socket_timeout`, `:max_retries`, `:sleep_between_retries_ms`, `:replica_policy`, `:read_mode_ap`, `:read_mode_sc`, `:read_touch_ttl_percent`, `:send_key`, `:use_compression`, and `:filter`. Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`. """ @spec exists(cluster(), Key.key_input(), read_opts()) :: {:ok, boolean()} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready | :no_master | :unknown_node} def exists(cluster, key, opts \\ []) do with {:ok, key} <- coerce_key(key) do Exists.execute(cluster, key, opts) end end @doc """ Updates `key`'s header metadata in `cluster`. Supported write opts are `:timeout`, `:socket_timeout`, `:max_retries`, `:sleep_between_retries_ms`, `:ttl`, `:generation`, `:generation_policy`, `:exists`, `:commit_level`, `:durable_delete`, `:respond_per_op`, `:send_key`, `:read_mode_ap`, `:read_mode_sc`, `:read_touch_ttl_percent`, `:use_compression`, and `:filter`. Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`. """ @spec touch(cluster(), Key.key_input(), write_opts()) :: {:ok, Aerospike.Record.metadata()} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready | :no_master | :unknown_node} def touch(cluster, key, opts \\ []) do with {:ok, key} <- coerce_key(key) do Touch.execute(cluster, key, opts) end end @doc """ Deletes `key` from `cluster`. Returns `{:ok, true}` when a record was deleted and `{:ok, false}` when the key was already absent. Supported write opts are `:timeout`, `:socket_timeout`, `:max_retries`, `:sleep_between_retries_ms`, `:ttl`, `:generation`, `:generation_policy`, `:exists`, `:commit_level`, `:durable_delete`, `:respond_per_op`, `:send_key`, `:read_mode_ap`, `:read_mode_sc`, `:read_touch_ttl_percent`, `:use_compression`, and `:filter`. Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`. """ @spec delete(cluster(), Key.key_input(), write_opts()) :: {:ok, boolean()} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready | :no_master | :unknown_node} def delete(cluster, key, opts \\ []) do with {:ok, key} <- coerce_key(key) do Delete.execute(cluster, key, opts) end end @doc """ Runs a constrained unary operate list for `key`. Supported operations: * `{:write, bin, value}` — simple bin write * `{:read, bin}` — simple bin read * `{:add, bin, delta}` — numeric increment * `{:append, bin, suffix}` — string suffix mutation * `{:prepend, bin, prefix}` — string prefix mutation * `:touch` — refresh record metadata * `:delete` — remove the record The command routes per input batch: read-only lists use read routing; any list that includes a write uses write routing. Supported opts are `:timeout`, `:socket_timeout`, `:max_retries`, `:sleep_between_retries_ms`, `:ttl`, `:generation`, `:generation_policy`, `:exists`, `:commit_level`, `:durable_delete`, `:respond_per_op`, `:send_key`, `:read_mode_ap`, `:read_mode_sc`, `:read_touch_ttl_percent`, `:use_compression`, and `:filter`. Accepted operations include the simple tuple form plus the public `Aerospike.Op` helpers for primitive, CDT, bit, HyperLogLog, and expression operations. Returned operation values are accumulated into `%Aerospike.Record{bins: map}` by bin name. Aerospike.operate(cluster, key, [ Aerospike.Op.Exp.read("projected", Aerospike.Exp.int_bin("count")), Aerospike.Op.Exp.write("computed", Aerospike.Exp.int(99)) ]) Aerospike.operate(cluster, key, [ Aerospike.Op.List.append("profile", "signed-in", ctx: [Aerospike.Ctx.map_key("events")] ) ]) Aerospike.operate(cluster, key, [ Aerospike.Op.Bit.count("flags", 0, 8), Aerospike.Op.HLL.get_count("visitors") ]) Accepts `%Aerospike.Key{}` or `{namespace, set, user_key}`. """ @spec operate(cluster(), Key.key_input(), [operate_operation()], write_opts()) :: {:ok, Aerospike.Record.t()} | {:error, Aerospike.Error.t()} | {:error, :cluster_not_ready | :no_master | :unknown_node} def operate(cluster, key, operations, opts \\ []) do with {:ok, key} <- coerce_key(key) do Operate.execute(cluster, key, operations, opts) end end defp coerce_key(key) do {:ok, Key.coerce!(key)} rescue err in ArgumentError -> {:error, Error.from_result_code(:invalid_argument, message: err.message)} end defp coerce_keys(keys) do {:ok, Enum.map(keys, &Key.coerce!/1)} rescue err in ArgumentError -> {:error, Error.from_result_code(:invalid_argument, message: err.message)} end defp batch_policy(cluster, opts) do cluster |> Cluster.retry_policy() |> Policy.batch(opts) end defp batch_command_entries(entries, %Policy.Batch{} = policy) do entries |> Enum.with_index() |> Enum.reduce_while({:ok, []}, &put_batch_command_entry(&1, &2, policy)) |> case do {:ok, command_entries} -> {:ok, Enum.reverse(command_entries)} {:error, %Error{}} = err -> err end end defp put_batch_command_entry({entry, index}, {:ok, acc}, %Policy.Batch{} = policy) do case batch_command_entry(entry, index, policy) do {:ok, %Entry{} = command_entry} -> {:cont, {:ok, [command_entry | acc]}} {:error, %Error{}} = err -> {:halt, err} end end defp batch_command_entry( %Batch.Read{key: %Key{} = key, opts: opts}, index, %Policy.Batch{} = policy ) do with {:ok, read_policy} <- Policy.batch_record_read(opts) do {:ok, %Entry{ index: index, key: key, kind: :read, dispatch: {:read, policy.retry.replica_policy, 0}, payload: batch_read_payload(read_policy) }} end end defp batch_command_entry( %Batch.Put{key: %Key{} = key, bins: bins, opts: opts}, index, %Policy.Batch{} ) when is_map(bins) do with {:ok, write_policy} <- Policy.batch_record_write(opts), {:ok, operations} <- batch_put_operations(bins) do {:ok, %Entry{ index: index, key: key, kind: :put, dispatch: :write, payload: Map.put(batch_write_payload(write_policy), :operations, operations) }} end end defp batch_command_entry(%Batch.Delete{key: %Key{} = key, opts: opts}, index, %Policy.Batch{}) do with {:ok, write_policy} <- Policy.batch_record_write(opts) do {:ok, %Entry{ index: index, key: key, kind: :delete, dispatch: :write, payload: batch_write_payload(write_policy) }} end end defp batch_command_entry( %Batch.Operate{key: %Key{} = key, operations: operations, opts: opts}, index, %Policy.Batch{} = policy ) when is_list(operations) do with :ok <- validate_batch_operations(operations) do flags = OperateFlags.scan_ops(operations) payload = batch_operate_payload(flags, opts) with {:ok, payload} <- payload do {:ok, %Entry{ index: index, key: key, kind: :operate, dispatch: batch_operate_dispatch(flags, policy), payload: Map.merge(payload, %{operations: operations, flags: flags}) }} end end end defp batch_command_entry( %Batch.UDF{ key: %Key{} = key, package: package, function: function, args: args, opts: opts }, index, %Policy.Batch{} ) when is_binary(package) and is_binary(function) and is_list(args) do with {:ok, write_policy} <- Policy.batch_record_write(opts) do {:ok, %Entry{ index: index, key: key, kind: :udf, dispatch: :write, payload: write_policy |> batch_write_payload() |> Map.merge(%{package: package, function: function, args: args}) }} end end defp batch_command_entry(_entry, _index, %Policy.Batch{}) do {:error, Error.from_result_code(:invalid_argument, message: "Aerospike.batch_operate/3 expects entries built by Aerospike.Batch" )} end defp batch_read_payload(%Policy.BatchRead{} = policy) do Map.take(policy, [ :filter, :read_mode_ap, :read_mode_sc, :read_touch_ttl_percent ]) end defp batch_write_payload(%Policy.BatchWrite{} = policy) do Map.take(policy, [ :ttl, :generation, :generation_policy, :filter, :exists, :commit_level, :durable_delete, :respond_per_op, :send_key, :read_mode_ap, :read_mode_sc, :read_touch_ttl_percent ]) end defp batch_operate_payload(%{has_write?: true}, opts) do with {:ok, write_policy} <- Policy.batch_record_write(opts) do {:ok, batch_write_payload(write_policy)} end end defp batch_operate_payload(%{has_write?: false}, opts) do with {:ok, read_policy} <- Policy.batch_record_read(opts) do {:ok, batch_read_payload(read_policy)} end end defp batch_operate_dispatch(%{has_write?: true}, %Policy.Batch{}), do: :write defp batch_operate_dispatch(%{has_write?: false}, %Policy.Batch{} = policy) do {:read, policy.retry.replica_policy, 0} end defp batch_put_operations(bins) when map_size(bins) == 0 do {:error, Error.from_result_code(:invalid_argument, message: "batch put requires at least one bin write" )} end defp batch_put_operations(bins) do bins |> Enum.reduce_while({:ok, []}, fn {bin_name, value}, {:ok, acc} -> case Operation.write(normalize_batch_bin_name(bin_name), value) do {:ok, operation} -> {:cont, {:ok, [operation | acc]}} {:error, %Error{}} = err -> {:halt, err} end end) |> case do {:ok, operations} -> {:ok, Enum.reverse(operations)} {:error, %Error{}} = err -> err end end defp validate_batch_operations([]) do {:error, Error.from_result_code(:invalid_argument, message: "batch operate requires at least one operation" )} end defp validate_batch_operations(operations) do if Enum.all?(operations, &match?(%Operation{}, &1)) do :ok else {:error, Error.from_result_code(:invalid_argument, message: "batch operate expects a list of Aerospike.Op operations" )} end end defp normalize_batch_bin_name(bin_name) when is_atom(bin_name), do: Atom.to_string(bin_name) defp normalize_batch_bin_name(bin_name), do: bin_name defp validate_truncate_opts(opts, callsite) when is_list(opts) do with :ok <- validate_truncate_opt_keys(opts, callsite), {:ok, _policy} <- Policy.admin_info(Keyword.delete(opts, :before)), :ok <- validate_truncate_before(opts) do {:ok, opts} end end defp validate_truncate_opt_keys([], _callsite), do: :ok defp validate_truncate_opt_keys([{key, _value} | rest], callsite) when key in [:before, :pool_checkout_timeout] do validate_truncate_opt_keys(rest, callsite) end defp validate_truncate_opt_keys([{key, _value} | _rest], callsite) do {:error, Error.from_result_code(:invalid_argument, message: "#{callsite} supports only :before and :pool_checkout_timeout options, got #{inspect(key)}" )} end defp validate_truncate_before(opts) do case Keyword.get(opts, :before) do nil -> :ok %DateTime{} -> :ok other -> {:error, Error.from_result_code(:invalid_argument, message: ":before must be a DateTime, got: #{inspect(other)}" )} end end defp validate_role_names(roles) when is_list(roles) do if Enum.all?(roles, &is_binary/1) do {:ok, roles} else {:error, Error.from_result_code(:invalid_argument, message: "roles must be a list of strings")} end end defp validate_privileges(privileges) when is_list(privileges) do if Enum.all?(privileges, &match?(%Privilege{}, &1)) do {:ok, privileges} else {:error, Error.from_result_code(:invalid_argument, message: "privileges must be a list of %Aerospike.Privilege{} structs" )} end end defp validate_create_role_opts(opts) when is_list(opts) do supported = [:whitelist, :read_quota, :write_quota, :timeout, :pool_checkout_timeout] with :ok <- validate_supported_opts(opts, supported, "Aerospike.create_role/4"), {:ok, whitelist} <- validate_role_whitelist(opts), {:ok, read_quota} <- validate_role_quota(opts, :read_quota), {:ok, write_quota} <- validate_role_quota(opts, :write_quota) do call_opts = Keyword.drop(opts, [:whitelist, :read_quota, :write_quota]) {:ok, {whitelist, read_quota, write_quota, call_opts}} end end defp validate_supported_opts(opts, supported, callsite) when is_list(opts) do case Enum.find(opts, fn {key, _value} -> key not in supported end) do nil -> :ok {key, _value} -> {:error, Error.from_result_code(:invalid_argument, message: "#{callsite} supports only #{format_supported_opts(supported)} options, got #{inspect(key)}" )} end end defp validate_role_whitelist(opts) do Keyword.get(opts, :whitelist, []) |> validate_role_whitelist_value() end defp validate_role_whitelist_value(whitelist) when is_list(whitelist) do if Enum.all?(whitelist, &is_binary/1) do {:ok, whitelist} else invalid_role_whitelist(whitelist) end end defp validate_role_whitelist_value(other), do: invalid_role_whitelist(other) defp validate_role_quota(opts, key) when key in [:read_quota, :write_quota] do opts |> Keyword.get(key, 0) |> validate_role_quota_value(key) end defp validate_role_quota_value(quota, _key) when is_integer(quota) and quota >= 0 do {:ok, quota} end defp validate_role_quota_value(other, key) when key in [:read_quota, :write_quota] do {:error, Error.from_result_code(:invalid_argument, message: ":#{key} must be a non-negative integer, got: #{inspect(other)}" )} end defp format_supported_opts([opt]), do: inspect(opt) defp format_supported_opts([left, right]), do: "#{inspect(left)} and #{inspect(right)}" defp format_supported_opts(opts) when is_list(opts) do {init, [last]} = Enum.split(opts, -1) Enum.map_join(init, ", ", &inspect/1) <> ", and " <> inspect(last) end defp invalid_role_whitelist(value) do {:error, Error.from_result_code(:invalid_argument, message: ":whitelist must be a list of strings, got: #{inspect(value)}" )} end defp validate_enable_metrics_opts(opts) when is_list(opts) do case Enum.find(opts, fn {key, _value} -> key != :reset end) do nil -> validate_enable_metrics_reset(opts) {key, _value} -> {:error, Error.from_result_code(:invalid_argument, message: "Aerospike.enable_metrics/2 supports only :reset option, got #{inspect(key)}" )} end end defp validate_enable_metrics_reset(opts) do case Keyword.get(opts, :reset, false) do reset when is_boolean(reset) -> :ok other -> {:error, Error.from_result_code(:invalid_argument, message: ":reset must be a boolean, got: #{inspect(other)}" )} end end defp validate_warm_up_opts(opts) when is_list(opts) do with :ok <- validate_supported_opts(opts, [:count, :pool_checkout_timeout], "Aerospike.warm_up/2"), :ok <- validate_non_neg_opt(opts, :count) do validate_non_neg_opt(opts, :pool_checkout_timeout) end end defp validate_non_neg_opt(opts, key) do case Keyword.fetch(opts, key) do :error -> :ok {:ok, value} when is_integer(value) and value >= 0 -> :ok {:ok, value} -> {:error, Error.from_result_code(:invalid_argument, message: ":#{key} must be a non-negative integer, got: #{inspect(value)}" )} end end end