fix(nif): harden OWN_GIL worker robustness

A per-request dict allocation failure in a subinterpreter worker no
longer breaks (and permanently kills) the worker command loop while
holding the GIL; it now sends an error response and keeps serving.

The owngil_* dispatch NIFs run on dirty IO schedulers and use the
non-blocking, deadline-bounded read_with_timeout/write_all_with_deadline
helpers (cmd_pipe write end set O_NONBLOCK), so a stalled or dead worker
can't wedge a scheduler forever or desync the framed protocol.

SuspensionRequired is now resolved from the current interpreter's erlang
module (like ProcessError), avoiding cross-interpreter PyObject use under
OWN_GIL.

Adds py_owngil_features_SUITE:owngil_reentrant_multi_stress_test.

Author: benoitc

CHANGELOG.md

@@ -4,6 +4,13 @@
 
 ### Security
 
+- **OWN_GIL worker robustness** (Python 3.14+) - A per-request allocation failure in
+  a subinterpreter worker no longer `break`s (and permanently kills) the worker command
+  loop; it returns an error and keeps serving. The `owngil_*` dispatch NIFs now run on
+  dirty IO schedulers and use non-blocking, deadline-bounded pipe reads and writes, so a
+  stalled or dead worker can't wedge a scheduler forever. The internal `SuspensionRequired`
+  exception is now looked up per-interpreter (like `ProcessError`), avoiding cross-
+  interpreter object use under OWN_GIL.
 - **Callback suspend/resume lifetime hardening** - The worker resource is now kept
   alive for the lifetime of a suspended callback (it could previously be GC'd mid-
   suspension, causing a use-after-free on resume). A resume frees any prior result

c_src/py_callback.c

@@ -147,6 +147,33 @@ static PyObject *get_current_process_error(void) {
     return exc_class;
 }
 
+/**
+ * Get the SuspensionRequired exception class from the current interpreter's
+ * erlang module. Under OWN_GIL subinterpreters each interpreter has its own
+ * erlang module/class, so raising the process-global object (which belongs to
+ * whichever interpreter initialized last) is cross-interpreter UB. Mirrors
+ * get_current_process_error().
+ */
+static PyObject *get_current_suspension_required(void) {
+    PyObject *erlang_module = PyImport_ImportModule("erlang");
+    if (erlang_module == NULL) {
+        PyErr_Clear();
+        return SuspensionRequiredException;  /* Fallback to global */
+    }
+
+    PyObject *exc_class = PyObject_GetAttrString(erlang_module, "SuspensionRequired");
+    Py_DECREF(erlang_module);
+
+    if (exc_class == NULL) {
+        PyErr_Clear();
+        return SuspensionRequiredException;  /* Fallback to global */
+    }
+
+    /* See get_current_process_error: decref and rely on the module keeping it alive. */
+    Py_DECREF(exc_class);
+    return exc_class;
+}
+
 /* ============================================================================
  * Callback Name Registry
  *
@@ -2117,7 +2144,7 @@ static PyObject *erlang_call_impl(PyObject *self, PyObject *args) {
     Py_XSETREF(tl_pending_args, call_args);
 
     /* Raise exception to abort Python execution */
-    PyErr_SetString(SuspensionRequiredException, "callback pending");
+    PyErr_SetString(get_current_suspension_required(), "callback pending");
     return NULL;
 }
 

c_src/py_nif.c

@@ -1724,6 +1724,10 @@ static ERL_NIF_TERM nif_set_callback_handler(ErlNifEnv *env, int argc, const ERL
  * dirty scheduler forever (the pipe write ends are set non-blocking). */
 #define CALLBACK_RESPONSE_IO_TIMEOUT_MS 30000
 
+/* Bound for OWN_GIL dispatch pipe I/O so a stalled/dead worker thread can't
+ * block the dispatching dirty scheduler forever. */
+#define OWNGIL_IO_TIMEOUT_MS 30000
+
 static ERL_NIF_TERM nif_send_callback_response(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]) {
     (void)argc;
     int fd;
@@ -7292,16 +7296,19 @@ static ERL_NIF_TERM nif_owngil_create_session(ErlNifEnv *env, int argc,
         .payload_len = 0,
     };
 
-    /* Write header */
-    if (write(w->cmd_pipe[1], &header, sizeof(header)) != sizeof(header)) {
+    /* Write header (non-blocking write end + deadline so a stalled/dead worker
+     * can't block this dirty scheduler forever). */
+    if (write_all_with_deadline(w->cmd_pipe[1], &header, sizeof(header),
+                                OWNGIL_IO_TIMEOUT_MS) != WRITE_OK) {
         pthread_mutex_unlock(&w->dispatch_mutex);
         return enif_make_tuple2(env, ATOM_ERROR,
                                 enif_make_atom(env, "write_failed"));
     }
 
-    /* Wait for response */
+    /* Wait for response, bounded by a deadline. */
     owngil_header_t resp;
-    if (read(w->result_pipe[0], &resp, sizeof(resp)) != sizeof(resp)) {
+    if (read_with_timeout(w->result_pipe[0], &resp, sizeof(resp),
+                          OWNGIL_IO_TIMEOUT_MS) != (ssize_t)sizeof(resp)) {
         pthread_mutex_unlock(&w->dispatch_mutex);
         return enif_make_tuple2(env, ATOM_ERROR,
                                 enif_make_atom(env, "read_failed"));
@@ -7385,9 +7392,11 @@ static ERL_NIF_TERM nif_owngil_submit_task(ErlNifEnv *env, int argc,
         .payload_len = payload_bin.size,
     };
 
-    /* Write header and payload */
-    if (write(w->cmd_pipe[1], &header, sizeof(header)) != sizeof(header) ||
-        write(w->cmd_pipe[1], payload_bin.data, payload_bin.size) != (ssize_t)payload_bin.size) {
+    /* Write header and payload (non-blocking write end + deadline). */
+    if (write_all_with_deadline(w->cmd_pipe[1], &header, sizeof(header),
+                                OWNGIL_IO_TIMEOUT_MS) != WRITE_OK ||
+        write_all_with_deadline(w->cmd_pipe[1], payload_bin.data, payload_bin.size,
+                                OWNGIL_IO_TIMEOUT_MS) != WRITE_OK) {
         pthread_mutex_unlock(&w->dispatch_mutex);
         enif_release_binary(&payload_bin);
         return enif_make_tuple2(env, ATOM_ERROR,
@@ -7443,11 +7452,13 @@ static ERL_NIF_TERM nif_owngil_destroy_session(ErlNifEnv *env, int argc,
         .payload_len = 0,
     };
 
-    /* Write header */
-    if (write(w->cmd_pipe[1], &header, sizeof(header)) == sizeof(header)) {
-        /* Wait for response */
+    /* Write header (best-effort, bounded). */
+    if (write_all_with_deadline(w->cmd_pipe[1], &header, sizeof(header),
+                                OWNGIL_IO_TIMEOUT_MS) == WRITE_OK) {
+        /* Wait for response (best-effort, bounded). */
         owngil_header_t resp;
-        read(w->result_pipe[0], &resp, sizeof(resp));
+        (void)read_with_timeout(w->result_pipe[0], &resp, sizeof(resp),
+                                OWNGIL_IO_TIMEOUT_MS);
     }
 
     pthread_mutex_unlock(&w->dispatch_mutex);
@@ -7504,12 +7515,15 @@ static ERL_NIF_TERM nif_owngil_apply_imports(ErlNifEnv *env, int argc,
         .payload_len = payload_bin.size,
     };
 
-    /* Write header and payload */
-    if (write(w->cmd_pipe[1], &header, sizeof(header)) == sizeof(header)) {
-        write(w->cmd_pipe[1], payload_bin.data, payload_bin.size);
-        /* Wait for response */
+    /* Write header and payload (best-effort, bounded). */
+    if (write_all_with_deadline(w->cmd_pipe[1], &header, sizeof(header),
+                                OWNGIL_IO_TIMEOUT_MS) == WRITE_OK) {
+        (void)write_all_with_deadline(w->cmd_pipe[1], payload_bin.data, payload_bin.size,
+                                      OWNGIL_IO_TIMEOUT_MS);
+        /* Wait for response (best-effort, bounded). */
         owngil_header_t resp;
-        read(w->result_pipe[0], &resp, sizeof(resp));
+        (void)read_with_timeout(w->result_pipe[0], &resp, sizeof(resp),
+                                OWNGIL_IO_TIMEOUT_MS);
     }
 
     enif_release_binary(&payload_bin);
@@ -7567,12 +7581,15 @@ static ERL_NIF_TERM nif_owngil_apply_paths(ErlNifEnv *env, int argc,
         .payload_len = payload_bin.size,
     };
 
-    /* Write header and payload */
-    if (write(w->cmd_pipe[1], &header, sizeof(header)) == sizeof(header)) {
-        write(w->cmd_pipe[1], payload_bin.data, payload_bin.size);
-        /* Wait for response */
+    /* Write header and payload (best-effort, bounded). */
+    if (write_all_with_deadline(w->cmd_pipe[1], &header, sizeof(header),
+                                OWNGIL_IO_TIMEOUT_MS) == WRITE_OK) {
+        (void)write_all_with_deadline(w->cmd_pipe[1], payload_bin.data, payload_bin.size,
+                                      OWNGIL_IO_TIMEOUT_MS);
+        /* Wait for response (best-effort, bounded). */
         owngil_header_t resp;
-        read(w->result_pipe[0], &resp, sizeof(resp));
+        (void)read_with_timeout(w->result_pipe[0], &resp, sizeof(resp),
+                                OWNGIL_IO_TIMEOUT_MS);
     }
 
     enif_release_binary(&payload_bin);
@@ -7866,11 +7883,11 @@ static ErlNifFunc nif_funcs[] = {
     {"subinterp_thread_pool_stats", 0, nif_subinterp_thread_pool_stats, 0},
 
     /* OWN_GIL session management for event loop pool */
-    {"owngil_create_session", 1, nif_owngil_create_session, 0},
-    {"owngil_submit_task", 7, nif_owngil_submit_task, 0},
-    {"owngil_destroy_session", 2, nif_owngil_destroy_session, 0},
-    {"owngil_apply_imports", 3, nif_owngil_apply_imports, 0},
-    {"owngil_apply_paths", 3, nif_owngil_apply_paths, 0},
+    {"owngil_create_session", 1, nif_owngil_create_session, ERL_NIF_DIRTY_JOB_IO_BOUND},
+    {"owngil_submit_task", 7, nif_owngil_submit_task, ERL_NIF_DIRTY_JOB_IO_BOUND},
+    {"owngil_destroy_session", 2, nif_owngil_destroy_session, ERL_NIF_DIRTY_JOB_IO_BOUND},
+    {"owngil_apply_imports", 3, nif_owngil_apply_imports, ERL_NIF_DIRTY_JOB_IO_BOUND},
+    {"owngil_apply_paths", 3, nif_owngil_apply_paths, ERL_NIF_DIRTY_JOB_IO_BOUND},
 
     /* Execution mode info */
     {"execution_mode", 0, nif_execution_mode, 0},

c_src/py_subinterp_thread.c

@@ -107,6 +107,13 @@ int subinterp_thread_pool_init(int num_workers) {
             pthread_mutex_destroy(&w->ns_mutex);
             goto cleanup_workers;
         }
+        /* Non-blocking cmd_pipe write end so the dispatch NIFs can bound their
+         * writes with write_all_with_deadline (no dirty-scheduler stall on a full
+         * pipe). The worker thread reads cmd_pipe[0] blocking, which is fine. */
+        {
+            int wfl = fcntl(w->cmd_pipe[1], F_GETFL, 0);
+            if (wfl >= 0) (void)fcntl(w->cmd_pipe[1], F_SETFL, wfl | O_NONBLOCK);
+        }
         if (pipe(w->result_pipe) < 0) {
             fprintf(stderr, "subinterp_thread_pool_init: failed to create result_pipe for worker %d: %s\n",
                     i, strerror(errno));
@@ -562,7 +569,12 @@ static void *worker_thread_main(void *arg) {
                     if ((owns_globals && globals == NULL) || (owns_locals && locals == NULL)) {
                         if (owns_globals) Py_XDECREF(globals);
                         if (owns_locals) Py_XDECREF(locals);
-                        break;
+                        /* Per-request dict allocation failure: respond with an
+                         * error and keep serving. This previously `break`ed the
+                         * worker command loop, permanently killing the thread (and
+                         * leaving the GIL held), wedging every session routed to it. */
+                        resp_header.msg_type = MSG_ERROR;
+                        goto send_response;
                     }
 
                     switch (header.req_type) {
@@ -933,6 +945,7 @@ static void *worker_thread_main(void *arg) {
             }
         }
 
+    send_response:
         if (tmp_env) {
             enif_free_env(tmp_env);
         }

test/py_owngil_features_SUITE.erl

@@ -47,6 +47,7 @@
     owngil_reentrant_basic_test/1,
     owngil_reentrant_nested_test/1,
     owngil_reentrant_concurrent_test/1,
+    owngil_reentrant_multi_stress_test/1,
     owngil_reentrant_complex_types_test/1,
     owngil_reentrant_thread_callback_test/1,
     owngil_reentrant_try_except_test/1
@@ -156,6 +157,7 @@ groups() ->
         owngil_reentrant_basic_test,
         owngil_reentrant_nested_test,
         owngil_reentrant_concurrent_test,
+        owngil_reentrant_multi_stress_test,
         owngil_reentrant_complex_types_test,
         owngil_reentrant_thread_callback_test,
         owngil_reentrant_try_except_test
@@ -705,6 +707,34 @@ owngil_reentrant_nested_test(_Config) ->
 
     py_context:stop(Ctx).
 
+%% @doc Stress the per-interpreter exception machinery (H8) and the bounded
+%% OWN_GIL dispatch (H7): many reentrant suspend/resume cycles across several
+%% subinterpreter contexts concurrently. A cross-interpreter exception object or
+%% a dispatch read/write desync would corrupt or crash the node under this load.
+owngil_reentrant_multi_stress_test(_Config) ->
+    py:register_function(owngil_level, fun([Level, N]) ->
+        case Level >= N of
+            true -> Level;
+            false ->
+                Code = iolist_to_binary(io_lib:format(
+                    "__import__('erlang').call('owngil_level', ~p, ~p)", [Level + 1, N])),
+                {ok, R} = py:eval(Code),
+                R
+        end
+    end),
+    Parent = self(),
+    Pids = [spawn(fun() ->
+        {ok, Ctx} = py_context:start_link(CtxId, owngil),
+        Ok = lists:all(fun(_) ->
+            {ok, 6} =:= py_context:eval(Ctx,
+                <<"__import__('erlang').call('owngil_level', 1, 6)">>, #{})
+        end, lists:seq(1, 20)),
+        py_context:stop(Ctx),
+        Parent ! {self(), Ok}
+    end) || CtxId <- lists:seq(1, 4)],
+    [receive {P, true} -> ok after 60000 -> error({stress_failed, P}) end || P <- Pids],
+    ok.
+
 %% @doc Concurrent callbacks from multiple owngil contexts
 owngil_reentrant_concurrent_test(_Config) ->
     NumContexts = 4,