Implement Cluster Scan (#113)

Signed-off-by: currantw <[email protected]>

Author: currantw

.gitignore

@@ -143,10 +143,11 @@ $RECYCLE.BIN/
 # Mac desktop service store files
 .DS_Store
 
-# IDE generaged files
+# IDE generated files
 .vs
 .vscode
 .kiro/
+.amazonq/
 
 _NCrunch*
 

rust/src/ffi.rs

@@ -317,7 +317,7 @@ pub(crate) unsafe fn create_route(
 /// * `data`, `data_len` and also each pointer stored in `data` must be able to be safely casted to a valid to a slice of the corresponding type via [`from_raw_parts`].
 ///   See the safety documentation of [`from_raw_parts`].
 /// * The caller is responsible of freeing the allocated memory.
-pub(crate) unsafe fn convert_double_pointer_to_vec<'a>(
+pub(crate) unsafe fn convert_string_pointer_array_to_vector<'a>(
     data: *const *const u8,
     len: usize,
     data_len: *const usize,
@@ -530,11 +530,11 @@ pub struct BatchOptionsInfo {
 /// * `cmd_ptr` must be able to be safely casted to a valid [`CmdInfo`]
 /// * `args` and `args_len` in a referred [`CmdInfo`] structure must not be `null`.
 /// * `data` in a referred [`CmdInfo`] structure must point to `arg_count` consecutive string pointers.
-/// * `args_len` in a referred [`CmdInfo`] structure must point to `arg_count` consecutive string lengths. See the safety documentation of [`convert_double_pointer_to_vec`].
+/// * `args_len` in a referred [`CmdInfo`] structure must point to `arg_count` consecutive string lengths. See the safety documentation of [`convert_string_pointer_array_to_vector`].
 pub(crate) unsafe fn create_cmd(ptr: *const CmdInfo) -> Result<Cmd, String> {
     let info = unsafe { *ptr };
     let arg_vec =
-        unsafe { convert_double_pointer_to_vec(info.args, info.arg_count, info.args_len) };
+        unsafe { convert_string_pointer_array_to_vector(info.args, info.arg_count, info.args_len) };
 
     let Some(mut cmd) = info.request_type.get_command() else {
         return Err("Couldn't fetch command type".into());

rust/src/lib.rs

@@ -11,6 +11,7 @@ use glide_core::{
 };
 use std::{
     ffi::{CStr, CString, c_char, c_void},
+    slice::from_raw_parts,
     sync::Arc,
 };
 use tokio::runtime::{Builder, Runtime};
@@ -438,3 +439,305 @@ pub unsafe extern "C" fn init(level: Option<Level>, file_name: *const c_char) ->
     let logger_level = logger_core::init(level.map(|level| level.into()), file_name_as_str);
     logger_level.into()
 }
+
+/// Execute a cluster scan request.
+///
+/// # Safety
+/// * `client_ptr` must be a valid Client pointer from create_client
+/// * `cursor` must be "0" for initial scan or a valid cursor ID from previous scan
+/// * `args` and `arg_lengths` must be valid arrays of length `arg_count`
+/// * `args` format: [b"MATCH", pattern_arg, b"COUNT", count, b"TYPE", type] (all optional)
+#[unsafe(no_mangle)]
+pub unsafe extern "C-unwind" fn request_cluster_scan(
+    client_ptr: *const c_void,
+    callback_index: usize,
+    cursor: *const c_char,
+    arg_count: u64,
+    args: *const usize,
+    arg_lengths: *const u64,
+) {
+    // Build client and add panic guard.
+    let client = unsafe {
+        Arc::increment_strong_count(client_ptr);
+        Arc::from_raw(client_ptr as *mut Client)
+    };
+    let core = client.core.clone();
+
+    let mut panic_guard = PanicGuard {
+        panicked: true,
+        failure_callback: core.failure_callback,
+        callback_index,
+    };
+
+    // Get the cluster scan state.
+    let cursor_id = unsafe { CStr::from_ptr(cursor) }
+        .to_str()
+        .unwrap_or("0")
+        .to_owned();
+
+    let scan_state_cursor = if cursor_id == "0" {
+        redis::ScanStateRC::new()
+    } else {
+        match glide_core::cluster_scan_container::get_cluster_scan_cursor(cursor_id.clone()) {
+            Ok(existing_cursor) => existing_cursor,
+            Err(_error) => {
+                unsafe {
+                    (core.failure_callback)(
+                        callback_index,
+                        format!("Invalid cursor ID: {}", cursor_id).as_ptr() as *const c_char,
+                        RequestErrorType::Unspecified,
+                    );
+                }
+                return;
+            }
+        }
+    };
+
+    // Build cluster scan arguments.
+    let cluster_scan_args = match unsafe {
+        build_cluster_scan_args(
+            arg_count,
+            args,
+            arg_lengths,
+            core.failure_callback,
+            callback_index,
+        )
+    } {
+        Some(args) => args,
+        None => return,
+    };
+
+    // Run cluster scan.
+    client.runtime.spawn(async move {
+        let mut async_panic_guard = PanicGuard {
+            panicked: true,
+            failure_callback: core.failure_callback,
+            callback_index,
+        };
+
+        let result = core
+            .client
+            .clone()
+            .cluster_scan(&scan_state_cursor, cluster_scan_args)
+            .await;
+        match result {
+            Ok(value) => {
+                let ptr = Box::into_raw(Box::new(ResponseValue::from_value(value)));
+                unsafe { (core.success_callback)(callback_index, ptr) };
+            }
+            Err(err) => unsafe {
+                report_error(
+                    core.failure_callback,
+                    callback_index,
+                    glide_core::errors::error_message(&err),
+                    glide_core::errors::error_type(&err),
+                );
+            },
+        };
+
+        async_panic_guard.panicked = false;
+    });
+
+    panic_guard.panicked = false;
+}
+
+/// Remove a cluster scan cursor from the Rust core container.
+///
+/// This should be called when the C# ClusterScanCursor is disposed or finalized
+/// to clean up resources allocated by the Rust core for cluster scan operations.
+///
+/// # Safety
+/// * `cursor_id` must be a valid C string or null
+#[unsafe(no_mangle)]
+pub unsafe extern "C" fn remove_cluster_scan_cursor(cursor_id: *const c_char) {
+    if cursor_id.is_null() {
+        return;
+    }
+
+    if let Ok(cursor_str) = unsafe { CStr::from_ptr(cursor_id).to_str() } {
+        glide_core::cluster_scan_container::remove_scan_state_cursor(cursor_str.to_string());
+    }
+}
+
+/// Build cluster scan arguments from C-style arrays.
+///
+/// # Arguments
+///
+/// * `arg_count` - The number of arguments in the arrays
+/// * `args` - Pointer to an array of pointers to argument data
+/// * `arg_lengths` - Pointer to an array of argument lengths
+/// * `failure_callback` - Callback function to invoke on error
+/// * `callback_index` - Index to pass to the callback function
+///
+/// # Safety
+/// * `args` and `arg_lengths` must be valid arrays of length `arg_count`
+/// * Each pointer in `args` must point to valid memory of the corresponding length
+unsafe fn build_cluster_scan_args(
+    arg_count: u64,
+    args: *const usize,
+    arg_lengths: *const u64,
+    failure_callback: FailureCallback,
+    callback_index: usize,
+) -> Option<redis::ClusterScanArgs> {
+    if arg_count == 0 {
+        return Some(redis::ClusterScanArgs::builder().build());
+    }
+
+    let arg_vec = unsafe { convert_string_pointer_array_to_vector(args, arg_count, arg_lengths) };
+
+    // Parse arguments from vector.
+    let mut pattern_arg: &[u8] = &[];
+    let mut type_arg: &[u8] = &[];
+    let mut count_arg: &[u8] = &[];
+
+    let mut iter = arg_vec.iter().peekable();
+    while let Some(arg) = iter.next() {
+        match *arg {
+            b"MATCH" => match iter.next() {
+                Some(p) => pattern_arg = p,
+                None => {
+                    unsafe {
+                        report_error(
+                            failure_callback,
+                            callback_index,
+                            "No argument following MATCH.".into(),
+                            RequestErrorType::Unspecified,
+                        );
+                    }
+                    return None;
+                }
+            },
+            b"TYPE" => match iter.next() {
+                Some(t) => type_arg = t,
+                None => {
+                    unsafe {
+                        report_error(
+                            failure_callback,
+                            callback_index,
+                            "No argument following TYPE.".into(),
+                            RequestErrorType::Unspecified,
+                        );
+                    }
+                    return None;
+                }
+            },
+            b"COUNT" => match iter.next() {
+                Some(c) => count_arg = c,
+                None => {
+                    unsafe {
+                        report_error(
+                            failure_callback,
+                            callback_index,
+                            "No argument following COUNT.".into(),
+                            RequestErrorType::Unspecified,
+                        );
+                    }
+                    return None;
+                }
+            },
+            _ => {
+                unsafe {
+                    report_error(
+                        failure_callback,
+                        callback_index,
+                        "Unknown cluster scan argument".into(),
+                        RequestErrorType::Unspecified,
+                    );
+                }
+                return None;
+            }
+        }
+    }
+
+    // Build cluster scan arguments.
+    let mut cluster_scan_args_builder = redis::ClusterScanArgs::builder();
+
+    if !pattern_arg.is_empty() {
+        cluster_scan_args_builder = cluster_scan_args_builder.with_match_pattern(pattern_arg);
+    }
+
+    if !type_arg.is_empty() {
+        let converted_type = match std::str::from_utf8(type_arg) {
+            Ok(t) => redis::ObjectType::from(t.to_string()),
+            Err(_) => {
+                unsafe {
+                    report_error(
+                        failure_callback,
+                        callback_index,
+                        "Invalid UTF-8 in TYPE argument".into(),
+                        RequestErrorType::Unspecified,
+                    );
+                }
+                return None;
+            }
+        };
+
+        cluster_scan_args_builder = cluster_scan_args_builder.with_object_type(converted_type);
+    }
+
+    if !count_arg.is_empty() {
+        let count_str = match std::str::from_utf8(count_arg) {
+            Ok(c) => c,
+            Err(_) => {
+                unsafe {
+                    report_error(
+                        failure_callback,
+                        callback_index,
+                        "Invalid UTF-8 in COUNT argument".into(),
+                        RequestErrorType::Unspecified,
+                    );
+                }
+                return None;
+            }
+        };
+
+        let converted_count = match count_str.parse::<u32>() {
+            Ok(c) => c,
+            Err(_) => {
+                unsafe {
+                    report_error(
+                        failure_callback,
+                        callback_index,
+                        "Invalid COUNT value".into(),
+                        RequestErrorType::Unspecified,
+                    );
+                }
+                return None;
+            }
+        };
+
+        cluster_scan_args_builder = cluster_scan_args_builder.with_count(converted_count);
+    }
+
+    Some(cluster_scan_args_builder.build())
+}
+
+/// Converts an array of pointers to strings to a vector of strings.
+///
+/// # Arguments
+///
+/// * `data` - Pointer to an array of pointers to string data
+/// * `len` - The number of strings in the array
+/// * `data_len` - Pointer to an array of string lengths
+///
+/// # Safety
+///
+/// `convert_string_pointer_array_to_vector` returns a `Vec` of u8 slice which holds pointers of C
+/// strings. The returned `Vec<&'a [u8]>` is meant to be copied into Rust code. Storing them
+/// for later use will cause the program to crash as the pointers will be freed by the caller.
+unsafe fn convert_string_pointer_array_to_vector<'a>(
+    data: *const usize,
+    len: u64,
+    data_len: *const u64,
+) -> Vec<&'a [u8]> {
+    let string_ptrs = unsafe { from_raw_parts(data, len as usize) };
+    let string_lengths = unsafe { from_raw_parts(data_len, len as usize) };
+
+    let mut result = Vec::<&[u8]>::with_capacity(string_ptrs.len());
+    for (i, &str_ptr) in string_ptrs.iter().enumerate() {
+        let slice = unsafe { from_raw_parts(str_ptr as *const u8, string_lengths[i] as usize) };
+        result.push(slice);
+    }
+
+    result
+}