fix: fix dv descriptor map insert potential FFI pointer leak (#3265)

## What changes are proposed in this pull request?

<!--
**Uncomment** this section if there are any changes affecting public
APIs. Else, **delete** this section.
### This PR affects the following public APIs
If there are breaking changes, please ensure the `breaking-changes`
label gets added by CI, and describe why the changes are needed.
Note that _new_ public APIs are not considered breaking.
-->

Currently, the rust -> FFI bridge -> caller contract is built on the
following assumption:

1. **Borrow:** Rust accesses the handle with `as_ref()` (`&T`) or
`as_mut()` (`& mut T`). The caller retains ownership and remains
responsible for passing the handle to its `free_*` function.
2. **Unconditional consume:** Rust calls `into_inner()` before any
fallible work. Rust owns the value from native entry onward and is
responsible for dropping it on every result, including errors. The
caller must not use or free the handle after the call.

The `dv_descriptor_map_insert` did not comply with either of these 2
above invariants, as it ran potentially errorsome checks against an
input string pointer before taking ownership of said pointer. This made
it difficult for FFI callers to know when it was safe to consider the
native memory fully handed-off to Rust.

This PR ensures `dv_descriptor_map_insert` takes full ownership of the
provided pointer before running path compliance checks so it remains
compliant with the second invariant above. The intended goal of this is
to make FFI integrations easier at the connector level, and pass more of
the implementation complexity onto core kernel.

## How was this change tested?

UTs

Author: jtuglu1

ffi/CLAUDE.md

@@ -17,6 +17,19 @@ options, etc.). Short-lived "plain old data" types like `ExternResult`, `KernelE
 
 Every handle has a corresponding `free_*` function (e.g. `free_engine`, `free_snapshot`).
 
+Handle parameters follow one of two ownership contracts:
+
+1. **Borrow:** Rust accesses the handle with `as_ref()` or `as_mut()`. The caller retains
+   ownership and remains responsible for passing the handle to its `free_*` function.
+2. **Unconditional consume:** Rust calls `into_inner()` before any fallible work. Rust owns the
+   value from native entry onward and is responsible for dropping it on every result, including
+   errors. The caller must not use or free the handle after the call.
+
+Do not conditionally consume a handle only when a fallible operation succeeds. Every function's
+safety documentation must state whether each handle is borrowed or consumed regardless of the
+result. For consuming functions, perform string parsing, visitor decoding, validation, and other
+fallible work only after all consumed handles have been converted with `into_inner()`.
+
 ## Error Handling
 
 Fallible functions return `ExternResult` (tagged union of Ok/Err). The caller provides an
@@ -166,9 +179,9 @@ transaction()
 
 The engine authors the DV file and passes descriptor fields to `dv_descriptor_new`. The
 descriptor map and scan iterator are both consumed by `transaction_update_deletion_vectors`;
-descriptor handles are consumed by `dv_descriptor_map_insert` only on success and must be
-freed by the caller on error. DV updates require both the `deletionVectors` reader/writer
-feature and `delta.enableDeletionVectors=true`.
+descriptor handles are consumed by `dv_descriptor_map_insert` regardless of the result. DV
+updates require both the `deletionVectors` reader/writer feature and
+`delta.enableDeletionVectors=true`.
 
 ## Tracing & Metrics
 

ffi/examples/update-dv/update_dv.c

@@ -163,12 +163,12 @@ int main(int argc, char* argv[]) {
 
   ExternResultbool insert_res =
       dv_descriptor_map_insert(map, data_file_path_slice, descriptor, engine);
+  descriptor = NULL; // consumed by dv_descriptor_map_insert regardless of result
   if (insert_res.tag != Okbool) {
     err = (Error*)insert_res.err;
     print_error("dv_descriptor_map_insert failed.", err);
     goto cleanup;
   }
-  descriptor = NULL; // consumed by dv_descriptor_map_insert on success
 
   // === Build a fresh scan metadata iterator for the update call ===
   ExternResultHandleMutableFfiSnapshotBuilder snapshot_builder_res =

ffi/src/transaction/deletion_vector.rs

@@ -57,8 +57,8 @@ pub unsafe extern "C" fn free_dv_descriptor_map(map: Handle<ExclusiveDvDescripto
     map.drop_handle();
 }
 
-/// Free a deletion vector descriptor handle. Only call this if the descriptor has not
-/// been moved into a map via [`dv_descriptor_map_insert`].
+/// Free a deletion vector descriptor handle. Only call this if the descriptor has not been
+/// consumed. [`dv_descriptor_map_insert`] consumes a descriptor handle regardless of its result.
 ///
 /// # Safety
 ///
@@ -173,9 +173,8 @@ fn dv_descriptor_new_impl(
 }
 
 /// Insert a deletion vector descriptor into the map under the given data file path.
-/// Consumes the descriptor handle on success. On error (e.g. invalid `data_file_path`),
-/// the descriptor handle is left untouched and must still be released by the caller via
-/// [`free_dv_descriptor`].
+/// Consumes the descriptor handle regardless of whether the insertion succeeds. On error (e.g.
+/// invalid `data_file_path`), the descriptor is dropped.
 ///
 /// `data_file_path` must be the data-file path exactly as it appears in the scan
 /// metadata produced by the kernel (the Add file action's `path` field). The kernel
@@ -185,7 +184,8 @@ fn dv_descriptor_new_impl(
 ///
 /// # Safety
 ///
-/// Caller must pass valid handles. The descriptor handle is consumed only on success.
+/// Caller must pass valid handles. The descriptor handle is consumed and must not be used or freed
+/// after this call, regardless of the result.
 #[no_mangle]
 pub unsafe extern "C" fn dv_descriptor_map_insert(
     mut map: Handle<ExclusiveDvDescriptorMap>,
@@ -195,24 +195,21 @@ pub unsafe extern "C" fn dv_descriptor_map_insert(
 ) -> ExternResult<bool> {
     let map_ref = unsafe { map.as_mut() };
     let engine_ref = unsafe { engine.as_ref() };
-    // Parse the path BEFORE taking ownership of the descriptor: if parsing fails the
-    // descriptor must remain valid so the caller can free it (otherwise we get a UAF
-    // when they retry or clean up).
+    let descriptor = unsafe { descriptor.into_inner() };
     let path_result = unsafe { TryFromStringSlice::try_from_slice(&data_file_path) };
-    dv_descriptor_map_insert_impl(map_ref, path_result, descriptor)
+    dv_descriptor_map_insert_impl(map_ref, path_result, *descriptor)
         .map(|_| true)
         .into_extern_result(&engine_ref)
 }
 
 fn dv_descriptor_map_insert_impl(
     map: &mut DvDescriptorMap,
     data_file_path: DeltaResult<&str>,
-    descriptor: Handle<ExclusiveDvDescriptor>,
+    descriptor: DeletionVectorDescriptor,
 ) -> DeltaResult<()> {
     let path = data_file_path?;
     let owned_path = path.to_string();
-    let descriptor = unsafe { descriptor.into_inner() };
-    map.inner.insert(owned_path, *descriptor);
+    map.inner.insert(owned_path, descriptor);
     Ok(())
 }
 
@@ -270,7 +267,28 @@ fn transaction_update_deletion_vectors_impl(
 
 #[cfg(test)]
 mod tests {
+    use std::sync::Arc;
+
+    use delta_kernel::Engine;
+
     use super::*;
+    use crate::error::{AllocateError, AllocateErrorFn, KernelError};
+    use crate::ffi_test_utils::{allocate_err, assert_extern_result_error_with_message};
+    use crate::{free_engine, ExternEngine};
+
+    struct ErrorOnlyExternEngine {
+        allocate_error: AllocateErrorFn,
+    }
+
+    impl ExternEngine for ErrorOnlyExternEngine {
+        fn engine(&self) -> Arc<dyn Engine> {
+            panic!("error-only test engine does not expose a kernel engine")
+        }
+
+        fn error_allocator(&self) -> &dyn AllocateError {
+            &self.allocate_error
+        }
+    }
 
     #[test]
     fn kernel_dv_storage_type_maps_to_kernel_storage_type() {
@@ -335,25 +353,57 @@ mod tests {
     }
 
     #[test]
-    fn dv_descriptor_map_insert_error_leaves_descriptor_freeable() {
+    fn dv_descriptor_map_insert_invalid_path_does_not_insert() {
         let mut map = DvDescriptorMap {
             inner: HashMap::new(),
         };
-        let descriptor =
+        let descriptor = unsafe {
             dv_descriptor_new_impl(KernelDvStorageType::Inline, Ok("ABC"), false, 0, 4, 1)
-                .expect("descriptor should be valid");
+                .expect("descriptor should be valid")
+                .into_inner()
+        };
 
         let result = dv_descriptor_map_insert_impl(
             &mut map,
             Err(Error::generic("bad data file path")),
-            descriptor.shallow_copy(),
+            *descriptor,
         );
 
-        assert!(
-            result.is_err(),
-            "insert should fail before consuming descriptor"
-        );
+        assert!(result.is_err(), "insert should fail for an invalid path");
         assert!(map.inner.is_empty());
-        unsafe { free_dv_descriptor(descriptor) };
+    }
+
+    #[test]
+    fn dv_descriptor_map_insert_invalid_utf8_consumes_descriptor() {
+        let engine: Arc<dyn ExternEngine> = Arc::new(ErrorOnlyExternEngine {
+            allocate_error: allocate_err,
+        });
+        let engine: Handle<SharedExternEngine> = engine.into();
+        let map = dv_descriptor_map_new();
+        let descriptor =
+            dv_descriptor_new_impl(KernelDvStorageType::Inline, Ok("ABC"), false, 0, 4, 1)
+                .expect("descriptor should be valid");
+        let invalid_utf8 = [0xff];
+        let invalid_path = KernelStringSlice {
+            ptr: invalid_utf8.as_ptr().cast(),
+            len: invalid_utf8.len(),
+        };
+
+        let result = unsafe {
+            dv_descriptor_map_insert(
+                map.shallow_copy(),
+                invalid_path,
+                descriptor,
+                engine.shallow_copy(),
+            )
+        };
+
+        assert_extern_result_error_with_message(result, KernelError::Utf8Error, None);
+        unsafe {
+            free_dv_descriptor_map(map);
+            free_engine(engine);
+        }
+        // The extern call consumed the descriptor on error. Deliberately having no descriptor
+        // cleanup here lets Miri's leak check catch consume-after-parse regressions.
     }
 }