From e6e09da9cb4f40bfb12f0f59c1be83ed02ec0e7b Mon Sep 17 00:00:00 2001 From: Shizuo Fujita Date: Fri, 14 Aug 2026 19:30:06 +0900 Subject: [PATCH 1/3] Fix UDF dispatch hang when an error message contains a NUL byte Callback errors reach DuckDB through set_error, which takes a C string. StringValueCStr ran outside rb_protect, so a message with an embedded NUL raised ArgumentError from inside the reporter itself. On a DuckDB worker thread that exception unwound the executor thread's loop before it signalled completion: the worker waited on done_cond forever, the executor thread was gone, and every later UDF callback in the process hung with it. Attacker- supplied row data interpolated into a raise message is enough to reach it. - read the message through rbduckdb_pending_error_message, which calls #message under rb_protect and replaces embedded NULs - signal the executor and proxy done flags from rb_ensure, and discard a StandardError a callback let escape, so one bad callback can no longer strand the dispatcher Co-Authored-By: Claude Opus 5 --- ext/duckdb/aggregate_function.c | 9 +- ext/duckdb/function_executor.c | 162 ++++++++++++++---- ext/duckdb/function_executor.h | 19 +- ext/duckdb/scalar_function.c | 18 +- ext/duckdb/table_function.c | 23 +-- .../function_error_message_test.rb | 91 ++++++++++ 6 files changed, 252 insertions(+), 70 deletions(-) create mode 100644 test/duckdb_test/function_error_message_test.rb diff --git a/ext/duckdb/aggregate_function.c b/ext/duckdb/aggregate_function.c index 5b7937da..590eadc8 100644 --- a/ext/duckdb/aggregate_function.c +++ b/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. */ diff --git a/ext/duckdb/function_executor.c b/ext/duckdb/function_executor.c index 0153eb8c..fe6060ac 100644 --- a/ext/duckdb/function_executor.c +++ b/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); + } } } diff --git a/ext/duckdb/function_executor.h b/ext/duckdb/function_executor.h index fded625f..b3790326 100644 --- a/ext/duckdb/function_executor.h +++ b/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); diff --git a/ext/duckdb/scalar_function.c b/ext/duckdb/scalar_function.c index c810bfcc..b6c1544b 100644 --- a/ext/duckdb/scalar_function.c +++ b/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); } } diff --git a/ext/duckdb/table_function.c b/ext/duckdb/table_function.c index 5b2d13dd..31a983ca 100644 --- a/ext/duckdb/table_function.c +++ b/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; diff --git a/test/duckdb_test/function_error_message_test.rb b/test/duckdb_test/function_error_message_test.rb new file mode 100644 index 00000000..695e73e2 --- /dev/null +++ b/test/duckdb_test/function_error_message_test.rb @@ -0,0 +1,91 @@ +# frozen_string_literal: true + +require 'test_helper' + +module DuckDBTest + # UDF errors reach DuckDB through C string entry points, so a message the C + # API cannot carry used to raise out of the reporter itself. On a DuckDB + # worker thread that killed the executor thread, and every later UDF callback + # waited for a completion signal that never came. + class FunctionErrorMessageTest < Minitest::Test + class MessageRaises < StandardError + def message + raise 'message itself raises' + end + end + + def setup + @db = DuckDB::Database.open + @con = @db.connect + # More than one row group, so the scan is parallel and the callbacks run + # on DuckDB worker threads rather than inline on the calling thread. + @con.query('CREATE TABLE t AS SELECT (i % 7)::VARCHAR AS a FROM range(300000) s(i)') + end + + def teardown + @con&.close + @db&.close + end + + def register_scalar(name, &block) + @con.register_scalar_function(DuckDB::ScalarFunction.new.tap do |sf| + sf.name = name + sf.add_parameter(DuckDB::LogicalType::VARCHAR) + sf.return_type = DuckDB::LogicalType::VARCHAR + sf.set_function(&block) + end) + end + + def assert_dispatch_alive(timeout = 60, &) + thread = Thread.new(&) + joined = thread.join(timeout) + thread.kill unless joined + + assert joined, 'UDF callback dispatch is stranded' + thread.value + end + + def test_scalar_error_message_with_a_null_byte_is_reported_to_duckdb + register_scalar('nul_boom') { |v| raise "bad token: \0#{v}" } + + error = assert_raises(DuckDB::Error) { @con.query('SELECT nul_boom(a) FROM t') } + + assert_match(/bad token:/, error.message) + end + + def test_scalar_error_message_that_cannot_be_read_is_reported_to_duckdb + register_scalar('unreadable_boom') { |_v| raise MessageRaises } + + assert_raises(DuckDB::Error) { @con.query('SELECT unreadable_boom(a) FROM t') } + end + + def test_aggregate_error_message_with_a_null_byte_is_reported_to_duckdb + @con.register_aggregate_function( + DuckDB::AggregateFunction.create( + name: 'nul_boom_agg', + return_type: :varchar, + params: [:varchar], + init: -> { +'' }, + update: ->(_state, v) { raise "bad token: \0#{v}" }, + combine: ->(state, _other) { state }, + finalize: ->(state) { state } + ) + ) + + error = assert_raises(DuckDB::Error) { @con.query('SELECT nul_boom_agg(a) FROM t') } + + assert_match(/bad token:/, error.message) + end + + def test_udf_dispatch_survives_an_error_message_duckdb_cannot_be_given + register_scalar('nul_boom2') { |v| raise "bad token: \0#{v}" } + register_scalar('echo') { |v| "echo#{v}" } + + assert_raises(StandardError) { @con.query('SELECT nul_boom2(a) FROM t') } + + result = assert_dispatch_alive { @con.query('SELECT echo(a) FROM t LIMIT 1').to_a } + + assert_equal [['echo0']], result + end + end +end From f0901ebbbaf8b45a88076afe1c2b5bc42aa66768 Mon Sep 17 00:00:00 2001 From: Shizuo Fujita Date: Fri, 14 Aug 2026 19:31:09 +0900 Subject: [PATCH 2/3] Add CHANGELOG entry for the UDF error reporter hang fix Co-Authored-By: Claude Opus 5 --- CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8efc73bb..2541b5ea 100644 --- a/CHANGELOG.md +++ b/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=`. From b0ebdb78f0c9d1a3b2ec462238ea49455e7152bc Mon Sep 17 00:00:00 2001 From: Shizuo Fujita Date: Fri, 14 Aug 2026 19:41:14 +0900 Subject: [PATCH 3/3] Bound each error-triggering query in the callback error tests A query that strands the dispatcher never returns, so the assertions around it could block the whole run before reaching the recovery check. Run each one on a thread with a join timeout and flunk instead. Co-Authored-By: Claude Opus 5 --- .../function_error_message_test.rb | 37 +++++++++++++------ 1 file changed, 26 insertions(+), 11 deletions(-) diff --git a/test/duckdb_test/function_error_message_test.rb b/test/duckdb_test/function_error_message_test.rb index 695e73e2..c9c9954a 100644 --- a/test/duckdb_test/function_error_message_test.rb +++ b/test/duckdb_test/function_error_message_test.rb @@ -36,19 +36,26 @@ def register_scalar(name, &block) end) end - def assert_dispatch_alive(timeout = 60, &) - thread = Thread.new(&) - joined = thread.join(timeout) - thread.kill unless joined + # Bounds a query that stops returning once the dispatcher is stranded, so a + # regression fails the run instead of blocking it forever. + def within_dispatch_timeout(timeout = 60, &block) + thread = Thread.new do + Thread.current.report_on_exception = false + block.call + end + return thread.value if thread.join(timeout) + + thread.kill - assert joined, 'UDF callback dispatch is stranded' - thread.value + flunk 'UDF callback dispatch is stranded' end def test_scalar_error_message_with_a_null_byte_is_reported_to_duckdb register_scalar('nul_boom') { |v| raise "bad token: \0#{v}" } - error = assert_raises(DuckDB::Error) { @con.query('SELECT nul_boom(a) FROM t') } + error = assert_raises(DuckDB::Error) do + within_dispatch_timeout { @con.query('SELECT nul_boom(a) FROM t') } + end assert_match(/bad token:/, error.message) end @@ -56,7 +63,9 @@ def test_scalar_error_message_with_a_null_byte_is_reported_to_duckdb def test_scalar_error_message_that_cannot_be_read_is_reported_to_duckdb register_scalar('unreadable_boom') { |_v| raise MessageRaises } - assert_raises(DuckDB::Error) { @con.query('SELECT unreadable_boom(a) FROM t') } + assert_raises(DuckDB::Error) do + within_dispatch_timeout { @con.query('SELECT unreadable_boom(a) FROM t') } + end end def test_aggregate_error_message_with_a_null_byte_is_reported_to_duckdb @@ -72,7 +81,9 @@ def test_aggregate_error_message_with_a_null_byte_is_reported_to_duckdb ) ) - error = assert_raises(DuckDB::Error) { @con.query('SELECT nul_boom_agg(a) FROM t') } + error = assert_raises(DuckDB::Error) do + within_dispatch_timeout { @con.query('SELECT nul_boom_agg(a) FROM t') } + end assert_match(/bad token:/, error.message) end @@ -81,9 +92,13 @@ def test_udf_dispatch_survives_an_error_message_duckdb_cannot_be_given register_scalar('nul_boom2') { |v| raise "bad token: \0#{v}" } register_scalar('echo') { |v| "echo#{v}" } - assert_raises(StandardError) { @con.query('SELECT nul_boom2(a) FROM t') } + # Deliberately loose: the reporting path is asserted above, and a regression + # there must not fail this test before it reaches the recovery check. + assert_raises(StandardError) do + within_dispatch_timeout { @con.query('SELECT nul_boom2(a) FROM t') } + end - result = assert_dispatch_alive { @con.query('SELECT echo(a) FROM t LIMIT 1').to_a } + result = within_dispatch_timeout { @con.query('SELECT echo(a) FROM t LIMIT 1').to_a } assert_equal [['echo0']], result end