aerospike-sync 3.0.0-alpha.2

// Copyright 2015-2020 Aerospike, Inc.
//
// Portions may be licensed to Aerospike, Inc. under one or more contributor
// license agreements.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not
// use this file except in compliance with the License. You may obtain a copy of
// the License at http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
// WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
// License for the specific language governing permissions and limitations under
// the License.

use std::future::Future;
use std::pin::Pin;
use std::str;
use std::sync::Arc;
#[cfg(feature = "rt-tokio")]
use std::sync::LazyLock;
use std::vec::Vec;

use crate::expressions::Expression;
use aerospike_core::errors::Result;
use aerospike_core::operations::{CdtContext, Operation};
use aerospike_core::query::PartitionFilter;
use aerospike_core::txn::{AbortStatus, CommitStatus, Txn};
use aerospike_core::DropIndexTask;
use aerospike_core::UdfRemoveTask;
use aerospike_core::{
    AdminPolicy, BatchOperation, BatchPolicy, BatchRecord, Bin, Bins, ClientPolicy,
    CollectionIndexType, ExecuteTask, IndexTask, IndexType, Key, Node, Privilege, QueryPolicy,
    ReadPolicy, Record, Recordset, RegisterTask, Role, Statement, ToHosts, TxnRollPolicy,
    TxnVerifyPolicy, UDFLang, User, Value, WritePolicy,
};
use futures::Stream;

#[cfg(feature = "rt-tokio")]
static SYNC_RT: LazyLock<tokio::runtime::Runtime> = LazyLock::new(|| {
    tokio::runtime::Builder::new_multi_thread()
        // Every blocking call funnels through this runtime, and the
        // client's background tasks (cluster tend, query workers) live
        // here too — give it a few workers, but don't take over the host.
        .worker_threads(
            std::thread::available_parallelism()
                .map(|n| n.get().min(4))
                .unwrap_or(1),
        )
        .enable_all()
        .thread_name("aerospike-sync-rt")
        .build()
        .expect("aerospike: failed to build sync runtime")
});

/// Drive a client future to completion from blocking code.
///
/// The future ALWAYS runs on the dedicated [`SYNC_RT`] runtime — never on
/// a caller's ambient runtime. This matters beyond style: client calls
/// spawn background tasks (cluster tend on connect, per-node query
/// workers, …), and tasks spawned while a future runs bind to the runtime
/// driving it. If that were a caller's runtime, those tasks would silently
/// die when the caller's runtime shuts down, poisoning the shared client
/// for everyone else.
#[cfg(feature = "rt-tokio")]
fn block_on<F>(f: F) -> F::Output
where
    F: Future + Send,
    F::Output: Send,
{
    use tokio::runtime::{Handle, RuntimeFlavor};

    match Handle::try_current() {
        // Inside a multi-thread runtime: blocking this worker thread is
        // permitted once marked with `block_in_place`; the future itself
        // still runs on SYNC_RT.
        Ok(h) if h.runtime_flavor() == RuntimeFlavor::MultiThread => {
            tokio::task::block_in_place(|| SYNC_RT.block_on(f))
        }
        // Inside a current-thread runtime: this thread may not block at
        // all (it IS the runtime), so hop to a scoped helper thread.
        Ok(_) => std::thread::scope(|s| {
            s.spawn(|| SYNC_RT.block_on(f))
                .join()
                .expect("aerospike sync bridge thread panicked")
        }),
        // Plain thread, no ambient runtime: block right here.
        Err(_) => SYNC_RT.block_on(f),
    }
}

/// Drive a client future to completion from blocking code.
///
/// Uses async-std's own `block_on` so the future runs on async-std's
/// global executor — the same one `aerospike_rt::spawn` targets. A
/// foreign executor (e.g. `futures::executor`) must not be used here: it
/// cannot drive async-std's IO and timers, so waiting on them would
/// deadlock or busy-spin.
#[cfg(feature = "rt-async-std")]
fn block_on<F: Future>(f: F) -> F::Output {
    async_std::task::block_on(f)
}

/// Blocking iterator over the items emitted by
/// [`Client::batch_stream`]. Each [`Iterator::next`] call returns the
/// next `(original_index, BatchRecord)` as soon as some per-node task
/// produces one; iteration ends once every per-node task has
/// finished and the underlying stream is closed.
///
/// Each `next()` drives the underlying stream through the crate's
/// [`block_on`] bridge: the calling thread parks with a real waker and
/// is woken exactly when an item (or the end of the stream) arrives —
/// no polling loop, no busy CPU while waiting.
pub struct BatchStream {
    inner: Pin<Box<dyn Stream<Item = (usize, BatchRecord)> + Send>>,
}

impl Iterator for BatchStream {
    type Item = (usize, BatchRecord);

    fn next(&mut self) -> Option<Self::Item> {
        use futures::StreamExt;
        block_on(self.inner.next())
    }
}

/// Instantiate a Client instance to access an Aerospike database cluster and perform database
/// operations.
///
/// The client is thread-safe. Only one client instance should be used per cluster. Multiple
/// threads should share this cluster instance.
///
/// Your application uses this class' API to perform database operations such as writing and
/// reading records, and selecting sets of records. Write operations include specialized
/// functionality such as append/prepend and arithmetic addition.
///
/// Each record may have multiple bins, unless the Aerospike server nodes are configured as
/// "single-bin". In "multi-bin" mode, partial records may be written or read by specifying the
/// relevant subset of bins.
pub struct Client {
    async_client: aerospike_core::Client,
}

unsafe impl Send for Client {}
unsafe impl Sync for Client {}

impl Client {
    /// Initializes Aerospike client with suitable hosts to seed the cluster map. The client policy
    /// is used to set defaults and size internal data structures. For each host connection that
    /// succeeds, the client will:
    ///
    /// - Add host to the cluster map
    /// - Request host's list of other nodes in cluster
    /// - Add these nodes to the cluster map
    ///
    /// In most cases, only one host is necessary to seed the cluster. The remaining hosts are
    /// added as future seeds in case of a complete network failure.
    ///
    /// If one connection succeeds, the client is ready to process database requests. If all
    /// connections fail and the policy's `fail_
    ///
    /// The seed hosts to connect to (one or more) can be specified as a comma-separated list of
    /// hostnames or IP addresses with optional port numbers, e.g.
    ///
    /// ```text
    /// 10.0.0.1:3000,10.0.0.2:3000,10.0.0.3:3000
    /// ```
    ///
    /// Port 3000 is used by default if the port number is omitted for any of the hosts.
    ///
    /// # Examples
    ///
    /// Using an environment variable to set the list of seed hosts.
    ///
    /// ```rust,edition2021
    /// use aerospike_sync::{Client, ClientPolicy};
    ///
    /// let hosts = std::env::var("AEROSPIKE_HOSTS").unwrap_or_else(|_| "127.0.0.1:3000".to_string());
    /// let client = Client::new(&ClientPolicy::default(), &hosts).unwrap();
    /// ```
    pub fn new(policy: &ClientPolicy, hosts: &(dyn ToHosts + Send + Sync)) -> Result<Self> {
        let client = block_on(aerospike_core::Client::new(policy, hosts))?;
        Ok(Client {
            async_client: client,
        })
    }

    /// Initializes the client like [`new`](Self::new) and attaches a dynamic
    /// configuration provider (e.g.
    /// [`YamlFileProvider`](aerospike_core::config::YamlFileProvider)) whose
    /// policy overrides are applied at runtime.
    #[cfg(feature = "dynamic-config")]
    pub fn new_with_config(
        policy: &ClientPolicy,
        hosts: &(dyn ToHosts + Send + Sync),
        provider: Arc<dyn aerospike_core::config::ConfigProvider>,
    ) -> Result<Self> {
        let client = block_on(aerospike_core::Client::new_with_config(
            policy, hosts, provider,
        ))?;
        Ok(Client {
            async_client: client,
        })
    }

    /// Closes the connection to the Aerospike cluster.
    pub fn close(&self) -> Result<()> {
        block_on(self.async_client.close())?;
        Ok(())
    }

    /// Returns `true` if the client is connected to any cluster nodes.
    pub fn is_connected(&self) -> bool {
        self.async_client.is_connected()
    }

    /// Returns this client library's version string (e.g. `"3.0.0-alpha.1"`).
    #[must_use]
    pub const fn client_version() -> &'static str {
        aerospike_core::Client::client_version()
    }

    /// Returns a list of the names of the active server nodes in the cluster.
    pub fn node_names(&self) -> Vec<String> {
        self.async_client.node_names()
    }

    /// Return node given its name.
    pub fn get_node(&self, name: &str) -> Result<Arc<Node>> {
        self.async_client.get_node(name)
    }

    /// Send info commands to a randomly selected cluster node and return
    /// the parsed key/value response. The map preserves the server's
    /// response order — one entry per command, in request order.
    pub fn info(
        &self,
        policy: &AdminPolicy,
        commands: &[&str],
    ) -> Result<aerospike_core::IndexMap<String, String>> {
        block_on(self.async_client.info(policy, commands))
    }


    /// Returns a list of active server nodes in the cluster.
    pub fn nodes(&self) -> Vec<Arc<Node>> {
        self.async_client.nodes()
    }

    /// Read record for the specified key. Depending on the bins value provided, all record bins,
    /// only selected record bins or only the record headers will be returned. The policy can be
    /// used to specify timeouts.
    ///
    /// # Examples
    ///
    /// Fetch specified bins for a record with the given key.
    ///
    /// ```rust,edition2021
    /// # use aerospike_sync::*;
    ///
    /// # let hosts = std::env::var("AEROSPIKE_HOSTS").unwrap_or_else(|_| "127.0.0.1:3000".to_string());
    /// # let client = Client::new(&ClientPolicy::default(), &hosts).unwrap();
    /// let key = as_key!("test", "test", "mykey");
    /// match client.get(&ReadPolicy::default(), &key, ["a", "b"]) {
    ///     Ok(record)
    ///         => println!("a={:?}", record.bins.get("a")),
    ///     Err(e) if e.server_result_code() == Some(ResultCode::KeyNotFoundError)
    ///         => println!("No such record: {}", key),
    ///     Err(err)
    ///         => println!("Error fetching record: {}", err),
    /// }
    /// ```
    ///
    /// Determine the remaining time-to-live of a record.
    ///
    /// ```rust,edition2021
    /// # use aerospike_sync::*;
    ///
    /// # let hosts = std::env::var("AEROSPIKE_HOSTS").unwrap_or_else(|_| "127.0.0.1:3000".to_string());
    /// # let client = Client::new(&ClientPolicy::default(), &hosts).unwrap();
    /// let key = as_key!("test", "test", "mykey");
    /// match client.get(&ReadPolicy::default(), &key, Bins::None) {
    ///     Ok(record) => {
    ///         match record.time_to_live() {
    ///             None => println!("record never expires"),
    ///             Some(duration) => println!("ttl: {} secs", duration.as_secs()),
    ///         }
    ///     },
    ///     Err(e) if e.server_result_code() == Some(ResultCode::KeyNotFoundError)
    ///         => println!("No such record: {}", key),
    ///     Err(err)
    ///         => println!("Error fetching record: {}", err),
    /// }
    /// ```
    ///
    /// # Panics
    /// Panics if the return is invalid
    pub fn get<T>(&self, policy: &ReadPolicy, key: &Key, bins: T) -> Result<Record>
    where
        T: Into<Bins> + Send + Sync + 'static,
    {
        block_on(self.async_client.get(policy, key, bins))
    }

    /// Read multiple record for specified batch keys in one batch call. This method allows
    /// different namespaces/bins to be requested for each key in the batch. If the `BatchRead` key
    /// field is not found, the corresponding record field will be `None`. The policy can be used
    /// to specify timeouts and maximum concurrent threads. This method requires Aerospike Server
    /// version >= 3.6.0.
    ///
    /// # Examples
    ///
    /// Fetch multiple records in a single client request
    ///
    /// ```rust,edition2021
    /// # use aerospike_sync::*;
    ///
    /// # let hosts = std::env::var("AEROSPIKE_HOSTS").unwrap_or_else(|_| "127.0.0.1:3000".to_string());
    /// # let client = Client::new(&ClientPolicy::default(), &hosts).unwrap();
    /// let bins = Bins::from(["name", "age"]);
    /// let bpr = BatchReadPolicy::default();
    /// let mut batch_ops = vec![];
    /// for i in 0..10 {
    ///   let key = as_key!("test", "test", i);
    ///   batch_ops.push(BatchOperation::read(&bpr, key, bins.clone()));
    /// }
    /// match client.batch(&BatchPolicy::default(), &batch_ops) {
    ///     Ok(results) => {
    ///       for result in results {
    ///         match result.record {
    ///           Some(record) => println!("{:?} => {:?}", result.key, record.bins),
    ///           None => println!("No such record: {:?}", result.key),
    ///         }
    ///       }
    ///     }
    ///     Err(err)
    ///         => println!("Error executing batch request: {}", err),
    /// }
    /// ```
    pub fn batch(
        &self,
        policy: &BatchPolicy,
        batch_records: &[BatchOperation],
    ) -> Result<Vec<BatchRecord>> {
        block_on(self.async_client.batch(policy, batch_records))
    }

    /// Execute multiple batch operations and return results as a
    /// blocking [`Iterator`]. Sync equivalent of the async
    /// `Client::batch_stream`.
    ///
    /// Unlike [`batch`](Self::batch), items arrive in the order they
    /// are received from each server node rather than the order of
    /// `ops`. Each item is `(original_index, BatchRecord)` so callers
    /// can match results back to their input vec. Iteration ends once
    /// every per-node task has completed.
    ///
    /// Internally this drives the async streaming variant: the
    /// per-node tasks are spawned on the underlying tokio runtime,
    /// each emits its `(idx, BatchRecord)` onto a channel as soon as
    /// it returns, and the [`BatchStream`] iterator pulls those items
    /// without ever blocking the runtime — it follows the same
    /// `try_recv` + `yield` pattern used by [`Recordset`].
    ///
    /// # Examples
    ///
    /// ```rust,no_run
    /// # use aerospike_sync::*;
    /// # let hosts = std::env::var("AEROSPIKE_HOSTS").unwrap_or_else(|_| "127.0.0.1:3000".to_string());
    /// # let client = Client::new(&ClientPolicy::default(), &hosts).unwrap();
    /// let key = as_key!("test", "test", 1);
    /// let batch = vec![BatchOperation::read(&BatchReadPolicy::default(), key, Bins::All)];
    /// for (idx, br) in client.batch_stream(&BatchPolicy::default(), batch).unwrap() {
    ///     println!("op[{}]: {:?}", idx, br.record);
    /// }
    /// ```
    pub fn batch_stream(
        &self,
        policy: &BatchPolicy,
        ops: Vec<BatchOperation>,
    ) -> Result<BatchStream> {
        let stream = block_on(self.async_client.batch_stream(policy, ops))?;
        Ok(BatchStream {
            inner: Box::pin(stream),
        })
    }

    /// Write record bin(s). The policy specifies the transaction timeout, record expiration and
    /// how the transaction is handled when the record already exists.
    ///
    /// # Examples
    ///
    /// Write a record with a single integer bin.
    ///
    /// ```rust,edition2021
    /// # use aerospike_sync::*;
    ///
    /// # let hosts = std::env::var("AEROSPIKE_HOSTS").unwrap_or_else(|_| "127.0.0.1:3000".to_string());
    /// # let client = Client::new(&ClientPolicy::default(), &hosts).unwrap();
    /// let key = as_key!("test", "test", "mykey");
    /// let bin = as_bin!("i", 42);
    /// match client.put(&WritePolicy::default(), &key, &vec![bin]) {
    ///     Ok(()) => println!("Record written"),
    ///     Err(err) => println!("Error writing record: {}", err),
    /// }
    /// ```
    ///
    /// Write a record with an expiration of 10 seconds.
    ///
    /// ```rust,edition2021
    /// # use aerospike_sync::*;
    ///
    /// # let hosts = std::env::var("AEROSPIKE_HOSTS").unwrap_or_else(|_| "127.0.0.1:3000".to_string());
    /// # let client = Client::new(&ClientPolicy::default(), &hosts).unwrap();
    /// let key = as_key!("test", "test", "mykey");
    /// let bin = as_bin!("i", 42);
    /// let mut policy = WritePolicy::default();
    /// policy.expiration = policy::Expiration::Seconds(10);
    /// match client.put(&policy, &key, &vec![bin]) {
    ///     Ok(()) => println!("Record written"),
    ///     Err(err) => println!("Error writing record: {}", err),
    /// }
    /// ```
    pub fn put<'a>(&self, policy: &'a WritePolicy, key: &'a Key, bins: &'a [Bin]) -> Result<()> {
        block_on(self.async_client.put(policy, key, bins))
    }

    /// Add integer bin values to existing record bin values. The policy specifies the transaction
    /// timeout, record expiration and how the transaction is handled when the record already
    /// exists. This call only works for integer values.
    ///
    /// # Examples
    ///
    /// Add two integer values to two existing bin values.
    ///
    /// ```rust,edition2021
    /// # use aerospike_sync::*;
    ///
    /// # let hosts = std::env::var("AEROSPIKE_HOSTS").unwrap_or_else(|_| "127.0.0.1:3000".to_string());
    /// # let client = Client::new(&ClientPolicy::default(), &hosts).unwrap();
    /// let key = as_key!("test", "test", "mykey");
    /// let bina = as_bin!("a", 1);
    /// let binb = as_bin!("b", 2);
    /// let bins = vec![bina, binb];
    /// match client.add(&WritePolicy::default(), &key, &bins) {
    ///     Ok(()) => println!("Record updated"),
    ///     Err(err) => println!("Error writing record: {}", err),
    /// }
    /// ```
    pub fn add<'a>(&self, policy: &'a WritePolicy, key: &'a Key, bins: &'a [Bin]) -> Result<()> {
        block_on(self.async_client.add(policy, key, bins))
    }

    /// Append bin string values to existing record bin values. The policy specifies the
    /// transaction timeout, record expiration and how the transaction is handled when the record
    /// already exists. This call only works for string values.
    pub fn append<'a>(&self, policy: &'a WritePolicy, key: &'a Key, bins: &'a [Bin]) -> Result<()> {
        block_on(self.async_client.append(policy, key, bins))
    }

    /// Prepend bin string values to existing record bin values. The policy specifies the
    /// transaction timeout, record expiration and how the transaction is handled when the record
    /// already exists. This call only works for string values.
    pub fn prepend<'a>(
        &self,
        policy: &'a WritePolicy,
        key: &'a Key,
        bins: &'a [Bin],
    ) -> Result<()> {
        block_on(self.async_client.prepend(policy, key, bins))
    }

    /// Delete record for specified key. The policy specifies the transaction timeout.
    /// The call returns `true` if the record existed on the server before deletion.
    ///
    /// # Examples
    ///
    /// Delete a record.
    ///
    /// ```rust,edition2021
    /// # use aerospike_sync::*;
    ///
    /// # let hosts = std::env::var("AEROSPIKE_HOSTS").unwrap_or_else(|_| "127.0.0.1:3000".to_string());
    /// # let client = Client::new(&ClientPolicy::default(), &hosts).unwrap();
    /// let key = as_key!("test", "test", "mykey");
    /// match client.delete(&WritePolicy::default(), &key) {
    ///     Ok(true) => println!("Record deleted"),
    ///     Ok(false) => println!("Record did not exist"),
    ///     Err(err) => println!("Error deleting record: {}", err),
    /// }
    /// ```
    pub fn delete(&self, policy: &WritePolicy, key: &Key) -> Result<bool> {
        block_on(self.async_client.delete(policy, key))
    }

    /// Reset record's time to expiration using the policy's expiration. Fail if the record does
    /// not exist.
    ///
    /// # Examples
    ///
    /// Reset a record's time to expiration to the default ttl for the namespace.
    ///
    /// ```rust,edition2021
    /// # use aerospike_sync::*;
    ///
    /// # let hosts = std::env::var("AEROSPIKE_HOSTS").unwrap_or_else(|_| "127.0.0.1:3000".to_string());
    /// # let client = Client::new(&ClientPolicy::default(), &hosts).unwrap();
    /// let key = as_key!("test", "test", "mykey");
    /// let mut policy = WritePolicy::default();
    /// policy.expiration = policy::Expiration::NamespaceDefault;
    /// match client.touch(&policy, &key) {
    ///     Ok(()) => println!("Record expiration updated"),
    ///     Err(err) => println!("Error writing record: {}", err),
    /// }
    /// ```
    pub fn touch(&self, policy: &WritePolicy, key: &Key) -> Result<()> {
        block_on(self.async_client.touch(policy, key))
    }

    /// Determine if a record key exists. The policy can be used to specify timeouts.
    pub fn exists(&self, policy: &ReadPolicy, key: &Key) -> Result<bool> {
        block_on(self.async_client.exists(policy, key))
    }

    /// Perform multiple read/write operations on a single key in one batch call.
    ///
    /// Operations on scalar values, lists and maps can be performed in the same call.
    ///
    /// Operations execute in the order specified by the client application.
    ///
    /// # Examples
    ///
    /// Add an integer value to an existing record and then read the result, all in one database
    /// call.
    ///
    /// ```rust,edition2021
    /// # use aerospike_sync::*;
    ///
    /// # let hosts = std::env::var("AEROSPIKE_HOSTS").unwrap_or_else(|_| "127.0.0.1:3000".to_string());
    /// # let client = Client::new(&ClientPolicy::default(), &hosts).unwrap();
    /// let key = as_key!("test", "test", "mykey");
    /// let bin = as_bin!("a", 42);
    /// let ops = vec![
    ///     operations::add(&bin),
    ///     operations::get_bin("a"),
    /// ];
    /// match client.operate(&WritePolicy::default(), &key, &ops) {
    ///     Ok(record) => println!("The new value is {}", record.bins.get("a").unwrap()),
    ///     Err(err) => println!("Error writing record: {}", err),
    /// }
    /// ```
    /// # Panics
    ///  Panics if the return is invalid
    pub fn operate(&self, policy: &WritePolicy, key: &Key, ops: &[Operation]) -> Result<Record> {
        block_on(self.async_client.operate(policy, key, ops))
    }

    /// Register a package containing user-defined functions (UDF) with the cluster. This
    /// asynchronous server call will return before the command is complete. The client registers
    /// the UDF package with a single, random cluster node; from there a copy will get distributed
    /// to all other cluster nodes automatically.
    ///
    /// Lua is the only supported scripting language for UDFs at the moment.
    ///
    /// # Examples
    ///
    /// ```rust,edition2021
    /// # use aerospike_sync::*;
    ///
    /// # let hosts = std::env::var("AEROSPIKE_HOSTS").unwrap_or_else(|_| "127.0.0.1:3000".to_string());
    /// # let client = Client::new(&ClientPolicy::default(), &hosts).unwrap();
    /// let code = r#"
    /// -- Validate value before writing.
    /// function writeWithValidation(r,name,value)
    ///   if (value >= 1 and value <= 10) then
    ///     if not aerospike:exists(r) then
    ///       aerospike:create(r)
    ///     end
    ///     r[name] = value
    ///     aerospike:update(r)
    ///   else
    ///       error("1000:Invalid value")
    ///   end
    /// end
    ///
    /// -- Set a particular bin only if record does not already exist.
    /// function writeUnique(r,name,value)
    ///   if not aerospike:exists(r) then
    ///     aerospike:create(r)
    ///     r[name] = value
    ///     aerospike:update(r)
    ///   end
    /// end
    /// "#;
    ///
    /// client.register_udf(&AdminPolicy::default(), code.as_bytes(),
    ///                     "example.lua", UDFLang::Lua).unwrap();
    /// ```
    pub fn register_udf(
        &self,
        policy: &AdminPolicy,
        udf_body: &[u8],
        server_path: &str,
        language: UDFLang,
    ) -> Result<RegisterTask> {
        block_on(
            self.async_client
                .register_udf(policy, udf_body, server_path, language),
        )
    }

    /// Register a package containing user-defined functions (UDF) with the cluster. This
    /// asynchronous server call will return before the command is complete. The client registers
    /// the UDF package with a single, random cluster node; from there a copy will get distributed
    /// to all other cluster nodes automatically.
    ///
    /// Lua is the only supported scripting language for UDFs at the moment.
    pub fn register_udf_from_file(
        &self,
        policy: &AdminPolicy,
        client_path: &str,
        server_path: &str,
        language: UDFLang,
    ) -> Result<RegisterTask> {
        block_on(self.async_client.register_udf_from_file(
            policy,
            client_path,
            server_path,
            language,
        ))
    }

    /// Remove a user-defined function (UDF) module from the server.
    pub fn remove_udf(&self, policy: &AdminPolicy, server_path: &str) -> Result<UdfRemoveTask> {
        block_on(self.async_client.remove_udf(policy, server_path))
    }

    /// Execute a user-defined function on the server and return the results. The function operates
    /// on a single record. The UDF package name is required to locate the UDF.
    ///
    /// # Panics
    /// Panics if the return is invalid
    pub fn execute_udf(
        &self,
        policy: &WritePolicy,
        key: &Key,
        server_path: &str,
        function_name: &str,
        args: Option<&[Value]>,
    ) -> Result<Option<Value>> {
        block_on(
            self.async_client
                .execute_udf(policy, key, server_path, function_name, args),
        )
    }

    /// Execute a query on all server nodes and return a record iterator. The query executor puts
    /// records on a queue in separate threads. The calling thread concurrently pops records off
    /// the queue through the record iterator.
    ///
    /// # Examples
    ///
    /// ```rust,edition2021
    /// # use aerospike_sync::*;
    ///
    /// # let hosts = std::env::var("AEROSPIKE_HOSTS").unwrap_or_else(|_| "127.0.0.1:3000".to_string());
    /// # let client = Client::new(&ClientPolicy::default(), &hosts).unwrap();
    /// let stmt = Statement::new("test", "test", Bins::All);
    /// let pf = PartitionFilter::all();
    /// match client.query(&QueryPolicy::default(), pf, stmt) {
    ///     Ok(records) => {
    ///         for record in &*records {
    ///             // .. process record
    ///         }
    ///     },
    ///     Err(err) => println!("Error fetching record: {}", err),
    /// }
    /// ```
    ///
    /// # Panics
    /// Panics if the async block fails
    pub fn query(
        &self,
        policy: &QueryPolicy,
        partition_filter: PartitionFilter,
        statement: Statement,
    ) -> Result<Arc<Recordset>> {
        block_on(self.async_client.query(policy, partition_filter, statement))
    }

    /// Execute a stream UDF aggregation query and return a
    /// [`aerospike_core::query::ResultSet`] with the aggregated values
    /// (requires the `lua` feature).
    ///
    /// The client-side portion of the UDF runs in an embedded Lua
    /// interpreter; the UDF source must be available locally — see
    /// [`aerospike_core::lua::set_lua_path`] and
    /// [`aerospike_core::lua::register_package`]. Iterate the returned set
    /// with its blocking `Iterator` impl (`for value in rs.as_ref() { … }`).
    #[cfg(feature = "lua")]
    pub fn query_aggregate(
        &self,
        policy: &QueryPolicy,
        statement: Statement,
        package_name: &str,
        function_name: &str,
        function_args: Option<&[Value]>,
    ) -> Result<Arc<aerospike_core::query::ResultSet>> {
        block_on(self.async_client.query_aggregate(
            policy,
            statement,
            package_name,
            function_name,
            function_args,
        ))
    }

    /// Execute a query and apply operations to matching records on the server.
    /// Returns an `ExecuteTask` that can be used to monitor the progress of the
    /// background job.
    pub fn query_operate(
        &self,
        write_policy: &WritePolicy,
        statement: Statement,
        operations: &[Operation],
    ) -> Result<ExecuteTask> {
        block_on(
            self.async_client
                .query_operate(write_policy, statement, operations),
        )
    }

    /// Apply a user-defined function to records matching the statement filter.
    /// Returns an `ExecuteTask` that can be used to monitor the progress of the
    /// background job.
    pub fn query_execute_udf(
        &self,
        write_policy: &WritePolicy,
        statement: Statement,
        package_name: &str,
        function_name: &str,
        args: Option<&[Value]>,
    ) -> Result<ExecuteTask> {
        block_on(self.async_client.query_execute_udf(
            write_policy,
            statement,
            package_name,
            function_name,
            args,
        ))
    }

    /// Sets XDR filter for given datacenter name and namespace. The expression filter indicates
    /// which records XDR should ship to the datacenter.
    /// Pass nil as filter to remove the current filter on the server.
    pub fn set_xdr_filter(
        &self,
        policy: &AdminPolicy,
        datacenter: &str,
        namespace: &str,
        filter_expression: Option<&Expression>,
    ) -> Result<()> {
        block_on(
            self.async_client
                .set_xdr_filter(policy, datacenter, namespace, filter_expression),
        )
    }

    /// Removes all records in the specified namespace/set efficiently.
    ///
    /// This method is many orders of magnitude faster than deleting records one at a time. It
    /// requires Aerospike Server version 3.12 or later. See
    /// <https://www.aerospike.com/docs/reference/info#truncate> for further info.
    ///
    /// The `set_name` is optional; set to `""` to delete all sets in `namespace`.
    ///
    /// `before_nanos` optionally specifies a last update timestamp (lut); if it is greater than
    /// zero, only records with a lut less than `before_nanos` are deleted. Units are in
    /// nanoseconds since unix epoch (1970-01-01). Pass in zero to delete all records in the
    /// namespace/set recardless of last update time.
    pub fn truncate(
        &self,
        policy: &AdminPolicy,
        namespace: &str,
        set_name: &str,
        before_nanos: i64,
    ) -> Result<()> {
        block_on(
            self.async_client
                .truncate(policy, namespace, set_name, before_nanos),
        )
    }

    /// Create a secondary index on a bin. This asynchronous server call
    /// returns before the command is complete.
    ///
    /// # Examples
    ///
    /// The following example creates an index `idx_foo_bar_baz`. The index is in namespace `foo`
    /// within set `bar` and bin `baz`:
    ///
    /// ```rust,edition2021
    /// # use aerospike_sync::*;
    ///
    /// # let hosts = std::env::var("AEROSPIKE_HOSTS").unwrap_or_else(|_| "127.0.0.1:3000".to_string());
    /// # let client = Client::new(&ClientPolicy::default(), &hosts).unwrap();
    /// let _ = client.create_index_on_bin(&AdminPolicy::default(), "foo", "bar", "baz",
    ///     "idx_foo_bar_baz", IndexType::Numeric, CollectionIndexType::Default, None);
    /// ```
    pub fn create_index_on_bin(
        &self,
        policy: &AdminPolicy,
        namespace: &str,
        set_name: &str,
        bin_name: &str,
        index_name: &str,
        index_type: IndexType,
        collection_index_type: CollectionIndexType,
        ctx: Option<&[CdtContext]>,
    ) -> Result<IndexTask> {
        block_on(self.async_client.create_index_on_bin(
            policy,
            namespace,
            set_name,
            bin_name,
            index_name,
            index_type,
            collection_index_type,
            ctx,
        ))
    }

    /// Create a secondary index using an expression. This asynchronous server call
    /// returns before the command is complete.
    ///
    /// # Examples
    ///
    /// The following example creates an index `idx_foo_bar_baz`. The index is in namespace `foo`
    /// within set `bar` with the expression `int_bin("a") == 500`:
    ///
    /// ```rust,edition2021
    /// # use aerospike_sync::*;
    /// # use aerospike_sync::expressions::{Expression, eq, int_bin, int_val};
    ///
    /// # let hosts = std::env::var("AEROSPIKE_HOSTS").unwrap_or_else(|_| "127.0.0.1:3000".to_string());
    /// # let client = Client::new(&ClientPolicy::default(), &hosts).unwrap();
    /// let fe: Expression = eq(int_bin("a".to_string()), int_val(500));
    /// let _ = client.create_index_using_expression(&AdminPolicy::default(), "foo", "bar",
    ///     "idx_foo_bar_baz", IndexType::Numeric, CollectionIndexType::Default, &fe);
    /// ```
    pub fn create_index_using_expression(
        &self,
        policy: &AdminPolicy,
        namespace: &str,
        set_name: &str,
        index_name: &str,
        index_type: IndexType,
        collection_index_type: CollectionIndexType,
        expression: &Expression,
    ) -> Result<IndexTask> {
        block_on(self.async_client.create_index_using_expression(
            policy,
            namespace,
            set_name,
            index_name,
            index_type,
            collection_index_type,
            expression,
        ))
    }

    /// Delete secondary index.
    pub fn drop_index(
        &self,
        policy: &AdminPolicy,
        namespace: &str,
        set_name: &str,
        index_name: &str,
    ) -> Result<DropIndexTask> {
        block_on(
            self.async_client
                .drop_index(policy, namespace, set_name, index_name),
        )
    }

    /// Creates a new user with password and roles. Clear-text password will be hashed using bcrypt
    /// before sending to server.
    pub fn create_user(
        &self,
        policy: &AdminPolicy,
        user: &str,
        password: &str,
        roles: &[&str],
    ) -> Result<()> {
        block_on(self.async_client.create_user(policy, user, password, roles))
    }

    /// Removes a user from the cluster.
    pub fn drop_user(&self, policy: &AdminPolicy, user: &str) -> Result<()> {
        block_on(self.async_client.drop_user(policy, user))
    }

    /// Changes a user's password. Clear-text password will be hashed using bcrypt before sending to server.
    pub fn change_password(&self, policy: &AdminPolicy, user: &str, password: &str) -> Result<()> {
        block_on(self.async_client.change_password(policy, user, password))
    }

    /// Adds roles to user's list of roles.
    pub fn grant_roles(&self, policy: &AdminPolicy, user: &str, roles: &[&str]) -> Result<()> {
        block_on(self.async_client.grant_roles(policy, user, roles))
    }

    /// Removes roles from user's list of roles.
    pub fn revoke_roles(&self, policy: &AdminPolicy, user: &str, roles: &[&str]) -> Result<()> {
        block_on(self.async_client.revoke_roles(policy, user, roles))
    }

    /// Retrieves users and their roles.
    /// If None is passed for the user argument, all users will be returned.
    pub fn query_users(&self, policy: &AdminPolicy, user: Option<&str>) -> Result<Vec<User>> {
        block_on(self.async_client.query_users(policy, user))
    }

    /// Creates a user-defined role.
    /// Quotas require server security configuration "enable-quotas" to be set to true.
    /// Pass 0 for quota values for no limit.
    pub fn create_role(
        &self,
        policy: &AdminPolicy,
        role_name: &str,
        privileges: &[Privilege],
        allowlist: &[&str],
        read_quota: u32,
        write_quota: u32,
    ) -> Result<()> {
        block_on(self.async_client.create_role(
            policy,
            role_name,
            privileges,
            allowlist,
            read_quota,
            write_quota,
        ))
    }

    /// Retrieves roles and their privileges.
    /// If None is passed for the role argument, all roles will be returned.
    pub fn query_roles(&self, policy: &AdminPolicy, role: Option<&str>) -> Result<Vec<Role>> {
        block_on(self.async_client.query_roles(policy, role))
    }

    /// Removes a user-defined role.
    pub fn drop_role(&self, policy: &AdminPolicy, role_name: &str) -> Result<()> {
        block_on(self.async_client.drop_role(policy, role_name))
    }

    /// Grants privileges to a user-defined role.
    pub fn grant_privileges(
        &self,
        policy: &AdminPolicy,
        role_name: &str,
        privileges: &[Privilege],
    ) -> Result<()> {
        block_on(
            self.async_client
                .grant_privileges(policy, role_name, privileges),
        )
    }

    /// Revokes privileges from a user-defined role.
    pub fn revoke_privileges(
        &self,
        policy: &AdminPolicy,
        role_name: &str,
        privileges: &[Privilege],
    ) -> Result<()> {
        block_on(
            self.async_client
                .revoke_privileges(policy, role_name, privileges),
        )
    }

    /// Sets IP address allowlist for a role.
    /// If allowlist is nil or empty, it removes existing allowlist from role.
    pub fn set_allowlist(
        &self,
        policy: &AdminPolicy,
        role_name: &str,
        allowlist: &[&str],
    ) -> Result<()> {
        block_on(
            self.async_client
                .set_allowlist(policy, role_name, allowlist),
        )
    }

    /// Sets maximum reads/writes per second limits for a role.
    /// If a quota is zero, the limit is removed.
    /// Quotas require server security configuration "enable-quotas" to be set to true.
    /// Pass 0 for quota values for no limit.
    pub fn set_quotas(
        &self,
        policy: &AdminPolicy,
        role_name: &str,
        read_quota: u32,
        write_quota: u32,
    ) -> Result<()> {
        block_on(
            self.async_client
                .set_quotas(policy, role_name, read_quota, write_quota),
        )
    }

    /// Creates a PKI user on the cluster: authentication is via client
    /// certificate, so no password is stored for the user.
    pub fn create_pki_user(&self, policy: &AdminPolicy, user: &str, roles: &[&str]) -> Result<()> {
        block_on(self.async_client.create_pki_user(policy, user, roles))
    }

    /// Commit a multi-record transaction (MRT) with default verify/roll
    /// policies. See [`aerospike_core::Client::commit`].
    pub fn commit(&self, txn: &Arc<Txn>) -> Result<CommitStatus> {
        block_on(self.async_client.commit(txn))
    }

    /// Commit a multi-record transaction with explicit verify and roll policies.
    pub fn commit_with_policies(
        &self,
        verify_policy: &TxnVerifyPolicy,
        roll_policy: &TxnRollPolicy,
        txn: &Arc<Txn>,
    ) -> Result<CommitStatus> {
        block_on(
            self.async_client
                .commit_with_policies(verify_policy, roll_policy, txn),
        )
    }

    /// Abort a multi-record transaction (MRT) with the default roll policy.
    /// See [`aerospike_core::Client::abort`].
    pub fn abort(&self, txn: &Arc<Txn>) -> Result<AbortStatus> {
        block_on(self.async_client.abort(txn))
    }

    /// Abort a multi-record transaction with an explicit roll policy.
    pub fn abort_with_policy(
        &self,
        roll_policy: &TxnRollPolicy,
        txn: &Arc<Txn>,
    ) -> Result<AbortStatus> {
        block_on(self.async_client.abort_with_policy(roll_policy, txn))
    }

    /// Enable metrics collection with the given policy.
    /// See [`aerospike_core::Client::enable_metrics`].
    pub fn enable_metrics(&self, policy: aerospike_core::MetricsPolicy) {
        self.async_client.enable_metrics(policy);
    }

    /// Disable metrics collection.
    pub fn disable_metrics(&self) {
        self.async_client.disable_metrics();
    }

    /// Whether metrics collection is currently enabled.
    pub fn metrics_enabled(&self) -> bool {
        self.async_client.metrics_enabled()
    }

    /// Snapshot of the cluster-wide metrics.
    pub fn metrics(&self) -> aerospike_core::ClusterMetrics {
        self.async_client.metrics()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::Duration;

    /// A BatchStream over a hand-built stream (no server, no items
    /// needed): the empty stream must terminate immediately and stay
    /// terminated.
    #[test]
    fn batch_stream_over_empty_stream_ends_immediately() {
        let mut bs = BatchStream {
            inner: Box::pin(futures::stream::iter(Vec::<(usize, BatchRecord)>::new())),
        };
        assert!(bs.next().is_none());
        assert!(bs.next().is_none());
    }

    /// The parked-consumer edge case: `next()` must genuinely park (not
    /// spin) and be woken when the producing side closes the channel.
    /// Uses an async_channel receiver as the stream — the same channel
    /// family the real batch executor uses.
    #[test]
    fn batch_stream_parked_next_wakes_when_producer_closes() {
        let (tx, rx) = async_channel::bounded::<(usize, BatchRecord)>(4);
        let mut bs = BatchStream { inner: Box::pin(rx) };

        let (done_tx, done_rx) = std::sync::mpsc::channel();
        let consumer = std::thread::spawn(move || {
            let item = bs.next(); // parks: channel empty, still open
            let _ = done_tx.send(item.is_none());
        });

        std::thread::sleep(Duration::from_millis(100));
        drop(tx); // producer goes away -> channel closes -> stream ends
        let ended_clean = done_rx
            .recv_timeout(Duration::from_secs(5))
            .expect("parked BatchStream::next was not woken by channel close");
        assert!(ended_clean, "expected None once the producer closed");
        consumer.join().unwrap();
    }
}