Fix UDF dispatch hang when an error message contains a NUL byte

CHANGELOG.md

@@ -3,6 +3,7 @@
 All notable changes to this project will be documented in this file.
 
 # Unreleased
+- fix a permanent hang when a UDF callback raised an exception whose message contained a NUL byte (or whose `#message` raised): the error reporter raised while reporting, killing the callback executor thread and stranding every later UDF callback in the process (PR #1448).
 - add `DuckDB::TableFunction::BindInfo#bind_data=` to store an arbitrary Ruby object as a custom table function's bind data.
 - add `DuckDB::TableFunction::FunctionInfo#bind_data` to retrieve, during execution, the object stored by `BindInfo#bind_data=`.
 - add `DuckDB::TableFunction::InitInfo#bind_data` to retrieve, during the init phase, the object stored by `BindInfo#bind_data=`.

ext/duckdb/aggregate_function.c

@@ -180,12 +180,9 @@ static inline void state_registry_remove(ruby_aggregate_state *state) {
  * Caller must only invoke this when rb_protect reported exception_state != 0.
  */
 static void report_ruby_error_to_duckdb(duckdb_function_info info) {
-    VALUE errinfo = rb_errinfo();
-    if (errinfo != Qnil) {
-        VALUE msg = rb_funcall(errinfo, rb_intern("message"), 0);
-        duckdb_aggregate_function_set_error(info, StringValueCStr(msg));
-    }
-    rb_set_errinfo(Qnil);
+    VALUE msg = rbduckdb_pending_error_message();
+    duckdb_aggregate_function_set_error(info, StringValueCStr(msg));
+    RB_GC_GUARD(msg);
 }
 
 /* state_size callback: constant buffer per state. */

ext/duckdb/function_executor.c

@@ -76,6 +76,52 @@ static int g_executor_started = 0;
  */
 static VALUE g_proxy_threads = Qnil;
 
+/*
+ * Replace embedded NUL bytes so the message can be handed to a C API.
+ * Returns str unchanged when there is nothing to replace.
+ */
+static VALUE message_without_nul(VALUE str) {
+    long len = RSTRING_LEN(str);
+    long i;
+    VALUE copy;
+    char *p;
+
+    if (memchr(RSTRING_PTR(str), '\0', (size_t)len) == NULL) {
+        return str;
+    }
+
+    copy = rb_str_new(RSTRING_PTR(str), len);
+    p = RSTRING_PTR(copy);
+    for (i = 0; i < len; i++) {
+        if (p[i] == '\0') {
+            p[i] = ' ';
+        }
+    }
+    return copy;
+}
+
+static VALUE pending_error_message(VALUE errinfo) {
+    return message_without_nul(rb_obj_as_string(rb_funcall(errinfo, rb_intern("message"), 0)));
+}
+
+VALUE rbduckdb_pending_error_message(void) {
+    VALUE errinfo = rb_errinfo();
+    VALUE msg;
+    int state = 0;
+
+    rb_set_errinfo(Qnil);
+    if (errinfo == Qnil) {
+        return rb_str_new_cstr("unknown error");
+    }
+
+    msg = rb_protect(pending_error_message, errinfo, &state);
+    if (state) {
+        rb_set_errinfo(Qnil);
+        return rb_str_new_cstr("an exception was raised, but its message could not be read");
+    }
+    return msg;
+}
+
 /* Data passed to the executor wait function */
 struct executor_wait_data {
     struct callback_request *request;
@@ -130,6 +176,49 @@ static void executor_stop_func(void *data) {
 #endif
 }
 
+/*
+ * Discard an exception a dispatched callback let escape, so that one bad
+ * callback cannot take a dispatcher thread down with it and strand every
+ * later callback. Interrupt, SystemExit and thread kill still get through.
+ */
+static void discard_callback_exception(int state) {
+    VALUE err = rb_errinfo();
+
+    if (err != Qnil && !rb_obj_is_kind_of(err, rb_eStandardError)) {
+        rb_jump_tag(state);
+    }
+    rb_set_errinfo(Qnil);
+}
+
+static VALUE request_run_body(VALUE varg) {
+    struct callback_request *req = (struct callback_request *)varg;
+    req->cb(req->user_data);
+    return Qnil;
+}
+
+/* Runs from rb_ensure: the worker waits on done_cond even if the callback raised. */
+static VALUE request_signal_done(VALUE varg) {
+    struct callback_request *req = (struct callback_request *)varg;
+
+#ifdef _MSC_VER
+    EnterCriticalSection(&req->done_lock);
+    req->done = 1;
+    WakeConditionVariable(&req->done_cond);
+    LeaveCriticalSection(&req->done_lock);
+#else
+    pthread_mutex_lock(&req->done_mutex);
+    req->done = 1;
+    pthread_cond_signal(&req->done_cond);
+    pthread_mutex_unlock(&req->done_mutex);
+#endif
+
+    return Qnil;
+}
+
+static VALUE request_run(VALUE varg) {
+    return rb_ensure(request_run_body, varg, request_signal_done, varg);
+}
+
 /* The executor thread main loop (Ruby thread) */
 static VALUE executor_thread_func(void *data) {
     struct executor_wait_data w;
@@ -140,23 +229,12 @@ static VALUE executor_thread_func(void *data) {
         rb_thread_call_without_gvl(executor_wait_func, &w, executor_stop_func, &w);
 
         if (w.request != NULL) {
-            struct callback_request *req = w.request;
-
-            /* Execute the callback with the GVL held */
-            req->cb(req->user_data);
+            int state = 0;
 
-            /* Signal the DuckDB worker thread that the callback is done */
-#ifdef _MSC_VER
-            EnterCriticalSection(&req->done_lock);
-            req->done = 1;
-            WakeConditionVariable(&req->done_cond);
-            LeaveCriticalSection(&req->done_lock);
-#else
-            pthread_mutex_lock(&req->done_mutex);
-            req->done = 1;
-            pthread_cond_signal(&req->done_cond);
-            pthread_mutex_unlock(&req->done_mutex);
-#endif
+            rb_protect(request_run, (VALUE)w.request, &state);
+            if (state) {
+                discard_callback_exception(state);
+            }
         }
     }
 
@@ -326,6 +404,37 @@ static void proxy_stop_func(void *data) {
 #endif
 }
 
+static VALUE proxy_run_body(VALUE data) {
+    struct worker_proxy *proxy = (struct worker_proxy *)data;
+    proxy->cb(proxy->user_data);
+    return Qnil;
+}
+
+/* Runs from rb_ensure: see request_signal_done. */
+static VALUE proxy_signal_done(VALUE data) {
+    struct worker_proxy *proxy = (struct worker_proxy *)data;
+
+#ifdef _MSC_VER
+    EnterCriticalSection(&proxy->lock);
+    proxy->has_request = 0;
+    proxy->request_done = 1;
+    WakeConditionVariable(&proxy->request_done_cond);
+    LeaveCriticalSection(&proxy->lock);
+#else
+    pthread_mutex_lock(&proxy->lock);
+    proxy->has_request = 0;
+    proxy->request_done = 1;
+    pthread_cond_signal(&proxy->request_done_cond);
+    pthread_mutex_unlock(&proxy->lock);
+#endif
+
+    return Qnil;
+}
+
+static VALUE proxy_run(VALUE data) {
+    return rb_ensure(proxy_run_body, data, proxy_signal_done, data);
+}
+
 /* The proxy thread main loop. Runs as the body of rb_ensure (see below). */
 static VALUE proxy_loop_body(VALUE data) {
     struct worker_proxy *proxy = (struct worker_proxy *)data;
@@ -337,23 +446,12 @@ static VALUE proxy_loop_body(VALUE data) {
         if (proxy->stop_requested) break;
 
         if (proxy->has_request) {
-            /* Execute the callback with the GVL held */
-            proxy->cb(proxy->user_data);
+            int state = 0;
 
-            /* Signal completion to the DuckDB worker thread */
-#ifdef _MSC_VER
-            EnterCriticalSection(&proxy->lock);
-            proxy->has_request = 0;
-            proxy->request_done = 1;
-            WakeConditionVariable(&proxy->request_done_cond);
-            LeaveCriticalSection(&proxy->lock);
-#else
-            pthread_mutex_lock(&proxy->lock);
-            proxy->has_request = 0;
-            proxy->request_done = 1;
-            pthread_cond_signal(&proxy->request_done_cond);
-            pthread_mutex_unlock(&proxy->lock);
-#endif
+            rb_protect(proxy_run, data, &state);
+            if (state) {
+                discard_callback_exception(state);
+            }
         }
     }
 

ext/duckdb/function_executor.h

@@ -18,8 +18,9 @@
  * A generic callback to be executed with the GVL held.
  * The user_data pointer is passed through unchanged from the dispatch call.
  *
- * The callback is responsible for catching Ruby exceptions (e.g. via
- * rb_protect) if needed — this module does not perform exception handling.
+ * The callback should still catch its own Ruby exceptions (e.g. via rb_protect)
+ * to report them to DuckDB. The dispatcher only guarantees that an exception it
+ * lets escape wakes the waiting worker instead of stranding it.
  */
 typedef void (*rbduckdb_function_callback_t)(void *user_data);
 
@@ -66,10 +67,8 @@ struct worker_proxy;
  * (typically by dispatching this through the global executor from a per-worker
  * init callback, which itself runs on a non-Ruby thread).
  *
- * May raise (NoMemError, Thread.new failure). The executor runs callbacks
- * unprotected, so a wrapper dispatched to it must rb_protect this call —
- * otherwise a raise longjmps past the executor's done-signaling and the
- * waiting DuckDB worker blocks forever.
+ * May raise (NoMemError, Thread.new failure), so a wrapper dispatched to the
+ * executor should rb_protect this call and fall back to the global executor.
  */
 struct worker_proxy *rbduckdb_worker_proxy_create(void);
 
@@ -87,6 +86,14 @@ void rbduckdb_worker_proxy_destroy(void *proxy);
  */
 void rbduckdb_function_executor_dispatch_via_proxy(rbduckdb_function_callback_t cb, void *user_data, struct worker_proxy *proxy);
 
+/*
+ * Take the pending Ruby exception's message as a String safe to hand to a
+ * DuckDB set_error entry point, and clear the exception. Always returns a
+ * String with no embedded NUL, so StringValueCStr on it cannot raise; #message
+ * runs under rb_protect, so neither can a hostile exception object.
+ */
+VALUE rbduckdb_pending_error_message(void);
+
 void *rbduckdb_function_data_register(VALUE value);
 VALUE rbduckdb_function_data_lookup(void *data);
 void rbduckdb_function_data_release(void *data);

ext/duckdb/scalar_function.c

@@ -70,12 +70,9 @@ static void execute_callback_protected(void *user_data) {
 
     rb_protect(execute_callback, (VALUE)arg, &exception_state);
     if (exception_state) {
-        VALUE errinfo = rb_errinfo();
-        if (errinfo != Qnil) {
-            VALUE msg = rb_funcall(errinfo, rb_intern("message"), 0);
-            duckdb_scalar_function_set_error(arg->info, StringValueCStr(msg));
-        }
-        rb_set_errinfo(Qnil);
+        VALUE msg = rbduckdb_pending_error_message();
+        duckdb_scalar_function_set_error(arg->info, StringValueCStr(msg));
+        RB_GC_GUARD(msg);
     }
 }
 
@@ -443,12 +440,9 @@ static void execute_bind_callback_protected(void *user_data) {
 
     rb_protect(call_bind_proc, (VALUE)&arg, &exception_state);
     if (exception_state) {
-        VALUE errinfo = rb_errinfo();
-        if (errinfo != Qnil) {
-            VALUE msg = rb_funcall(errinfo, rb_intern("message"), 0);
-            duckdb_scalar_function_bind_set_error(darg->info, StringValueCStr(msg));
-        }
-        rb_set_errinfo(Qnil);
+        VALUE msg = rbduckdb_pending_error_message();
+        duckdb_scalar_function_bind_set_error(darg->info, StringValueCStr(msg));
+        RB_GC_GUARD(msg);
     }
 }
 

ext/duckdb/table_function.c

@@ -243,10 +243,9 @@ static void execute_bind_callback_protected(void *user_data) {
     rb_protect(call_bind_proc, (VALUE)call_args, &state);
 
     if (state) {
-        VALUE err = rb_errinfo();
-        VALUE msg = rb_funcall(err, rb_intern("message"), 0);
+        VALUE msg = rbduckdb_pending_error_message();
         duckdb_bind_set_error(darg->info, StringValueCStr(msg));
-        rb_set_errinfo(Qnil);
+        RB_GC_GUARD(msg);
     }
 }
 
@@ -319,10 +318,9 @@ static void execute_init_callback_protected(void *user_data) {
     rb_protect(call_init_proc, (VALUE)call_args, &state);
 
     if (state) {
-        VALUE err = rb_errinfo();
-        VALUE msg = rb_funcall(err, rb_intern("message"), 0);
+        VALUE msg = rbduckdb_pending_error_message();
         duckdb_init_set_error(darg->info, StringValueCStr(msg));
-        rb_set_errinfo(Qnil);
+        RB_GC_GUARD(msg);
     }
 }
 
@@ -404,10 +402,9 @@ static void execute_execute_callback_protected(void *user_data) {
     rb_protect(call_execute_proc, (VALUE)call_args, &state);
 
     if (state) {
-        VALUE err = rb_errinfo();
-        VALUE msg = rb_funcall(err, rb_intern("message"), 0);
+        VALUE msg = rbduckdb_pending_error_message();
         duckdb_function_set_error(darg->info, StringValueCStr(msg));
-        rb_set_errinfo(Qnil);
+        RB_GC_GUARD(msg);
     }
 }
 
@@ -452,11 +449,9 @@ static VALUE create_proxy_callback(VALUE varg) {
 }
 
 /*
- * rbduckdb_worker_proxy_create may raise (NoMemError, Thread.new failure),
- * and the executor runs callbacks unprotected — a raise would longjmp past
- * its done-signaling and block the waiting DuckDB worker forever. Swallow
- * the exception instead: the proxy stays NULL, local_init sets no state, and
- * the execute callback falls back to the global executor.
+ * rbduckdb_worker_proxy_create may raise (NoMemError, Thread.new failure).
+ * Swallow it: the proxy stays NULL, local_init sets no state, and the execute
+ * callback falls back to the global executor.
  */
 static void create_proxy_callback_protected(void *user_data) {
     int exception_state;