clickhouse-c-rs: improve safety

clickhouse-c-rs/README.md

@@ -51,10 +51,11 @@ same allocator the C side used. `Client` boxes its `Allocator` so the
 heap address the C side stashes in `c->al` stays valid through every
 later call & through `chc_client_close`.
 
-**No-copy column slabs.** `chc_block_builder_append_*` retains raw
-pointers to caller-owned bytes until `chc_block_write`. Mirrored as
-`BlockBuilder<'a>`; each `append_*` takes `&'a [u8]` / `&'a [u64]` &
-each appended `TypeRef<'a>`. Caller keeps slabs alive for `'a`.
+**No-copy columns.** `chc_block_builder_append_*` retains raw pointers
+to caller-owned names and bytes for the builder lifetime. Mirrored as
+`BlockBuilder<'a>`; each `append_*` takes `&'a str`, `&'a [u8]` /
+`&'a [u64]` & each appended `TypeRef<'a>`. Caller keeps inputs alive
+for `'a`.
 
 **Self-referential C structs.** `chc_io` carries a pointer back into the
 `chc_posix_io` state it was initialized from; `PosixIo` holds both inline
@@ -193,6 +194,8 @@ let mut opts = ClientOpts::new()
 opts.compression = Compression::Lz4;
 
 let mut client = Client::init(&opts, Allocator::stdlib(), io, Some(codec))?;
+// Refresh before each blocking operation to apply a fresh absolute deadline:
+// client.set_read_timeout(Some(std::time::Duration::from_secs(30)))?;
 
 client.send_query("INSERT INTO t FORMAT Native", None)?;
 // send one or more data blocks via client.send_data(Some(&bb)),

clickhouse-c-rs/src/alloc.rs

@@ -19,7 +19,7 @@ impl Allocator {
     /// travels through the vtable's `ud` slot, so the allocator must outlive
     /// every object parsed through it, hence `'static`. Alignment is fixed at
     /// `align_of::<u128>()`, the max_align_t that `stdlib()`'s malloc gives.
-    pub fn global<A: GlobalAlloc>(a: &'static A) -> Self {
+    pub fn global<A: GlobalAlloc + Sync>(a: &'static A) -> Self {
         Self {
             raw: sys::chc_alloc {
                 ud: a as *const A as *mut c_void,
@@ -53,31 +53,154 @@ mod vtable {
     // matches the max_align_t that stdlib()'s malloc guarantees.
     const ALIGN: usize = core::mem::align_of::<u128>();
 
-    // ALIGN is a fixed power of two; sizes here never approach isize::MAX.
     #[inline]
-    fn layout(bytes: usize) -> Layout {
-        unsafe { Layout::from_size_align_unchecked(bytes, ALIGN) }
+    fn layout(bytes: usize) -> Option<Layout> {
+        Layout::from_size_align(bytes.max(1), ALIGN).ok()
     }
 
-    pub extern "C" fn alloc<A: GlobalAlloc>(ud: *mut c_void, bytes: usize) -> *mut c_void {
+    pub extern "C" fn alloc<A: GlobalAlloc + Sync>(ud: *mut c_void, bytes: usize) -> *mut c_void {
         let a = unsafe { &*ud.cast::<A>() };
-        unsafe { a.alloc(layout(bytes)).cast() }
+        let Some(layout) = layout(bytes) else {
+            return core::ptr::null_mut();
+        };
+        unsafe { a.alloc(layout).cast() }
     }
 
-    // GlobalAlloc::realloc reads only the alignment from the old layout; the
-    // allocator tracks the block size internally.
-    pub extern "C" fn realloc<A: GlobalAlloc>(
+    pub extern "C" fn realloc<A: GlobalAlloc + Sync>(
         ud: *mut c_void,
         p: *mut c_void,
-        _old_bytes: usize,
+        old_bytes: usize,
         new_bytes: usize,
     ) -> *mut c_void {
+        if p.is_null() {
+            return alloc::<A>(ud, new_bytes);
+        }
         let a = unsafe { &*ud.cast::<A>() };
-        unsafe { a.realloc(p.cast(), layout(0), new_bytes).cast() }
+        let Some(old_layout) = layout(old_bytes) else {
+            return core::ptr::null_mut();
+        };
+        if new_bytes == 0 {
+            unsafe { a.dealloc(p.cast(), old_layout) };
+            return core::ptr::null_mut();
+        }
+        unsafe { a.realloc(p.cast(), old_layout, new_bytes).cast() }
     }
 
-    pub extern "C" fn free<A: GlobalAlloc>(ud: *mut c_void, p: *mut c_void, _bytes: usize) {
+    pub extern "C" fn free<A: GlobalAlloc + Sync>(ud: *mut c_void, p: *mut c_void, bytes: usize) {
+        if p.is_null() {
+            return;
+        }
         let a = unsafe { &*ud.cast::<A>() };
-        unsafe { a.dealloc(p.cast(), layout(0)) }
+        let Some(layout) = layout(bytes) else {
+            return;
+        };
+        unsafe { a.dealloc(p.cast(), layout) }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use core::alloc::{GlobalAlloc, Layout};
+    use core::sync::atomic::{AtomicBool, Ordering};
+    use std::alloc::System;
+    use std::collections::HashMap;
+    use std::sync::{LazyLock, Mutex};
+
+    use super::Allocator;
+    use crate::{BlockBuilder, TypeAst};
+
+    static CHECKED: LazyLock<CheckedAlloc> = LazyLock::new(|| CheckedAlloc {
+        live: Mutex::new(HashMap::new()),
+        invalid_layout: AtomicBool::new(false),
+    });
+
+    struct CheckedAlloc {
+        live: Mutex<HashMap<usize, Layout>>,
+        invalid_layout: AtomicBool,
+    }
+
+    unsafe impl GlobalAlloc for CheckedAlloc {
+        unsafe fn alloc(&self, layout: Layout) -> *mut u8 {
+            let ptr = unsafe { System.alloc(layout) };
+            if !ptr.is_null() {
+                self.live
+                    .lock()
+                    .expect("checked allocator lock")
+                    .insert(ptr as usize, layout);
+            }
+            ptr
+        }
+
+        unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
+            let Some(actual) = self
+                .live
+                .lock()
+                .expect("checked allocator lock")
+                .remove(&(ptr as usize))
+            else {
+                self.invalid_layout.store(true, Ordering::Relaxed);
+                return;
+            };
+            if actual != layout {
+                self.invalid_layout.store(true, Ordering::Relaxed);
+            }
+            unsafe { System.dealloc(ptr, actual) };
+        }
+
+        unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
+            let Some(actual) = self
+                .live
+                .lock()
+                .expect("checked allocator lock")
+                .remove(&(ptr as usize))
+            else {
+                self.invalid_layout.store(true, Ordering::Relaxed);
+                return core::ptr::null_mut();
+            };
+            if actual != layout {
+                self.invalid_layout.store(true, Ordering::Relaxed);
+            }
+            let new_ptr = unsafe { System.realloc(ptr, actual, new_size) };
+            let mut live = self.live.lock().expect("checked allocator lock");
+            if new_ptr.is_null() {
+                live.insert(ptr as usize, actual);
+            } else {
+                let new_layout = Layout::from_size_align(new_size, actual.align()).expect("layout");
+                live.insert(new_ptr as usize, new_layout);
+            }
+            new_ptr
+        }
+    }
+
+    #[test]
+    fn global_allocator_preserves_layouts() {
+        CHECKED.invalid_layout.store(false, Ordering::Relaxed);
+        assert!(
+            CHECKED
+                .live
+                .lock()
+                .expect("checked allocator lock")
+                .is_empty()
+        );
+
+        let alloc = Allocator::global(&*CHECKED);
+        drop(BlockBuilder::new(alloc).expect("empty builder"));
+        let ty = TypeAst::parse("UInt32", alloc).expect("UInt32");
+        let data = 7u32.to_le_bytes();
+        let mut builder = BlockBuilder::new(alloc).expect("builder");
+        builder
+            .append_fixed("x", ty.view(), &data, 1)
+            .expect("append");
+        drop(builder);
+        drop(ty);
+
+        assert!(!CHECKED.invalid_layout.load(Ordering::Relaxed));
+        assert!(
+            CHECKED
+                .live
+                .lock()
+                .expect("checked allocator lock")
+                .is_empty()
+        );
     }
 }

clickhouse-c-rs/src/async_client.rs

@@ -133,6 +133,7 @@ impl AsyncClient {
         opts: ClientOpts,
         codec: Option<Pin<Box<Codec>>>,
     ) -> Result<Self> {
+        opts.validate_codec(codec.as_ref().map(|codec| codec.as_ref()))?;
         let alloc = Box::new(Allocator::stdlib());
         let read_buf_bytes = if opts.read_buffer_bytes == 0 {
             DEFAULT_READ_BUF_BYTES