Use flag-based callback detection instead of exception type

Previously, erlang.call() raised SuspensionRequired and the executor
detected it by checking the exception type. This broke when ASGI/WSGI
frameworks caught the exception before it reached the executor.

Now we set a thread-local flag (tl_pending_callback) that the executor
checks FIRST, before looking at exception type. The exception is still
raised to abort Python execution, but we don't care if Python code
catches/re-raises it - the flag tells us a callback is pending.

This makes erlang.call() work reliably from any Python context,
including ASGI middleware that catches all exceptions.

Author: benoitc

c_src/py_callback.c

@@ -702,51 +702,44 @@ static PyObject *erlang_call_impl(PyObject *self, PyObject *args) {
     }
 
     /*
-     * Suspension is allowed - raise SuspensionRequired exception.
+     * Flag-based suspension: set thread-local flag and raise exception.
      *
-     * Unlike returning a marker tuple, raising an exception properly interrupts
-     * Python execution even in the middle of an expression like:
-     *   erlang.call('foo', x) + 1
+     * Unlike checking exception type (which fails if frameworks catch exceptions),
+     * we set a thread-local flag that the C executor checks FIRST. This way:
+     * 1. Python code can catch/re-raise the exception - we don't care
+     * 2. The flag tells us a callback is pending
+     * 3. Executor handles it before looking at exception type
      *
-     * The executor (process_request) catches this exception and:
-     * 1. Creates a suspended state resource
-     * 2. Returns {suspended, CallbackId, StateRef, {Func, Args}} to Erlang
-     * 3. The dirty scheduler is freed
-     *
-     * When Erlang calls resume_callback with the result:
-     * 1. The result is stored in the suspended state
-     * 2. A dirty NIF is scheduled to re-run the Python code
-     * 3. On re-run, this function finds the cached result and returns it
+     * The exception is just to abort Python execution cleanly.
      */
     uint64_t callback_id = atomic_fetch_add(&g_callback_id_counter, 1);
 
-    /* Create exception args tuple: (callback_id, func_name, args) */
-    PyObject *exc_args = PyTuple_New(3);
-    if (exc_args == NULL) {
-        Py_DECREF(call_args);
-        return NULL;
-    }
-
-    PyObject *callback_id_obj = PyLong_FromUnsignedLongLong(callback_id);
-    PyObject *func_name_obj = PyUnicode_FromString(func_name);
+    /* Set pending callback flag and store info */
+    tl_pending_callback = true;
+    tl_pending_callback_id = callback_id;
 
-    if (callback_id_obj == NULL || func_name_obj == NULL) {
-        Py_XDECREF(callback_id_obj);
-        Py_XDECREF(func_name_obj);
+    /* Store function name (make a copy) */
+    if (tl_pending_func_name != NULL) {
+        enif_free(tl_pending_func_name);
+    }
+    tl_pending_func_name = enif_alloc(func_name_len + 1);
+    if (tl_pending_func_name == NULL) {
+        tl_pending_callback = false;
         Py_DECREF(call_args);
-        Py_DECREF(exc_args);
+        PyErr_SetString(PyExc_MemoryError, "Failed to allocate function name");
         return NULL;
     }
+    memcpy(tl_pending_func_name, func_name, func_name_len);
+    tl_pending_func_name[func_name_len] = '\0';
+    tl_pending_func_name_len = func_name_len;
 
-    PyTuple_SET_ITEM(exc_args, 0, callback_id_obj); /* Takes ownership */
-    PyTuple_SET_ITEM(exc_args, 1, func_name_obj);   /* Takes ownership */
-    PyTuple_SET_ITEM(exc_args, 2, call_args);       /* Takes ownership */
+    /* Store args (take ownership) */
+    Py_XDECREF(tl_pending_args);
+    tl_pending_args = call_args;  /* Takes ownership, don't decref */
 
-    /* Raise the exception - Python will unwind the stack */
-    PyErr_SetObject(SuspensionRequiredException, exc_args);
-    Py_DECREF(exc_args);  /* SetObject increfs, so we decref our reference */
-
-    return NULL;  /* Signals exception was raised */
+    /* Raise exception to abort Python execution */
+    PyErr_SetString(SuspensionRequiredException, "callback pending");
+    return NULL;
 }
 
 /**

c_src/py_exec.c

@@ -290,50 +290,65 @@ static void process_request(py_request_t *req) {
             if (check_timeout_error()) {
                 PyErr_Clear();
                 req->result = enif_make_tuple2(env, ATOM_ERROR, ATOM_TIMEOUT);
-            } else if (is_suspension_exception()) {
+            } else if (tl_pending_callback) {
                 /*
-                 * Python code called erlang.call() which raised SuspensionRequired.
-                 * Create a suspended state and return {suspended, CallbackId, StateRef, {Func, Args}}
-                 * so Erlang can execute the callback and then resume.
+                 * Flag-based callback detection: check flag FIRST, not exception type.
+                 * This works even if Python code caught and re-raised the exception.
                  */
-                PyObject *exc_args = get_suspension_args();  /* Clears exception */
+                PyErr_Clear();  /* Clear whatever exception is set */
+
+                /* Build exc_args tuple from thread-local storage */
+                PyObject *exc_args = PyTuple_New(3);
                 if (exc_args == NULL) {
-                    req->result = make_error(env, "get_suspension_args_failed");
+                    tl_pending_callback = false;
+                    req->result = make_error(env, "alloc_exc_args_failed");
                 } else {
-                    suspended_state_t *suspended = create_suspended_state(env, exc_args, req);
-                    if (suspended == NULL) {
+                    PyObject *callback_id_obj = PyLong_FromUnsignedLongLong(tl_pending_callback_id);
+                    PyObject *func_name_obj = PyUnicode_FromStringAndSize(
+                        tl_pending_func_name, tl_pending_func_name_len);
+
+                    if (callback_id_obj == NULL || func_name_obj == NULL) {
+                        Py_XDECREF(callback_id_obj);
+                        Py_XDECREF(func_name_obj);
                         Py_DECREF(exc_args);
-                        req->result = make_error(env, "create_suspended_state_failed");
+                        tl_pending_callback = false;
+                        req->result = make_error(env, "build_exc_args_failed");
                     } else {
-                        /* Extract callback info from exception args */
-                        PyObject *callback_id_obj = PyTuple_GetItem(exc_args, 0);
-                        PyObject *func_name_obj = PyTuple_GetItem(exc_args, 1);
-                        PyObject *call_args_obj = PyTuple_GetItem(exc_args, 2);
+                        PyTuple_SET_ITEM(exc_args, 0, callback_id_obj);
+                        PyTuple_SET_ITEM(exc_args, 1, func_name_obj);
+                        Py_INCREF(tl_pending_args);  /* Tuple takes ownership */
+                        PyTuple_SET_ITEM(exc_args, 2, tl_pending_args);
 
-                        uint64_t callback_id = PyLong_AsUnsignedLongLong(callback_id_obj);
-                        Py_ssize_t fn_len;
-                        const char *fn = PyUnicode_AsUTF8AndSize(func_name_obj, &fn_len);
+                        suspended_state_t *suspended = create_suspended_state(env, exc_args, req);
+                        if (suspended == NULL) {
+                            Py_DECREF(exc_args);
+                            tl_pending_callback = false;
+                            req->result = make_error(env, "create_suspended_state_failed");
+                        } else {
+                            /* Create Erlang terms */
+                            ERL_NIF_TERM state_ref = enif_make_resource(env, suspended);
+                            enif_release_resource(suspended);
 
-                        /* Create Erlang terms */
-                        ERL_NIF_TERM state_ref = enif_make_resource(env, suspended);
-                        enif_release_resource(suspended);  /* Erlang now holds the reference */
+                            ERL_NIF_TERM callback_id_term = enif_make_uint64(env, tl_pending_callback_id);
 
-                        ERL_NIF_TERM callback_id_term = enif_make_uint64(env, callback_id);
+                            ERL_NIF_TERM func_name_term;
+                            unsigned char *fn_buf = enif_make_new_binary(env, tl_pending_func_name_len, &func_name_term);
+                            memcpy(fn_buf, tl_pending_func_name, tl_pending_func_name_len);
 
-                        ERL_NIF_TERM func_name_term;
-                        unsigned char *fn_buf = enif_make_new_binary(env, fn_len, &func_name_term);
-                        memcpy(fn_buf, fn, fn_len);
+                            ERL_NIF_TERM args_term = py_to_term(env, tl_pending_args);
 
-                        ERL_NIF_TERM args_term = py_to_term(env, call_args_obj);
+                            Py_DECREF(exc_args);
 
-                        Py_DECREF(exc_args);
+                            /* Clear pending state */
+                            tl_pending_callback = false;
 
-                        /* Return {suspended, CallbackId, StateRef, {FuncName, Args}} */
-                        req->result = enif_make_tuple4(env,
-                            ATOM_SUSPENDED,
-                            callback_id_term,
-                            state_ref,
-                            enif_make_tuple2(env, func_name_term, args_term));
+                            /* Return {suspended, CallbackId, StateRef, {FuncName, Args}} */
+                            req->result = enif_make_tuple4(env,
+                                ATOM_SUSPENDED,
+                                callback_id_term,
+                                state_ref,
+                                enif_make_tuple2(env, func_name_term, args_term));
+                        }
                     }
                 }
             } else {
@@ -404,43 +419,57 @@ static void process_request(py_request_t *req) {
                 if (check_timeout_error()) {
                     PyErr_Clear();
                     req->result = enif_make_tuple2(env, ATOM_ERROR, ATOM_TIMEOUT);
-                } else if (is_suspension_exception()) {
-                    /* Handle suspension from erlang.call() in eval */
-                    PyObject *exc_args = get_suspension_args();  /* Clears exception */
+                } else if (tl_pending_callback) {
+                    /* Flag-based callback detection for eval */
+                    PyErr_Clear();
+
+                    PyObject *exc_args = PyTuple_New(3);
                     if (exc_args == NULL) {
-                        req->result = make_error(env, "get_suspension_args_failed");
+                        tl_pending_callback = false;
+                        req->result = make_error(env, "alloc_exc_args_failed");
                     } else {
-                        suspended_state_t *suspended = create_suspended_state(env, exc_args, req);
-                        if (suspended == NULL) {
-                            Py_DECREF(exc_args);
-                            req->result = make_error(env, "create_suspended_state_failed");
-                        } else {
-                            PyObject *callback_id_obj = PyTuple_GetItem(exc_args, 0);
-                            PyObject *func_name_obj = PyTuple_GetItem(exc_args, 1);
-                            PyObject *call_args_obj = PyTuple_GetItem(exc_args, 2);
-
-                            uint64_t callback_id = PyLong_AsUnsignedLongLong(callback_id_obj);
-                            Py_ssize_t fn_len;
-                            const char *fn = PyUnicode_AsUTF8AndSize(func_name_obj, &fn_len);
-
-                            ERL_NIF_TERM state_ref = enif_make_resource(env, suspended);
-                            enif_release_resource(suspended);
-
-                            ERL_NIF_TERM callback_id_term = enif_make_uint64(env, callback_id);
-
-                            ERL_NIF_TERM func_name_term;
-                            unsigned char *fn_buf = enif_make_new_binary(env, fn_len, &func_name_term);
-                            memcpy(fn_buf, fn, fn_len);
-
-                            ERL_NIF_TERM args_term = py_to_term(env, call_args_obj);
+                        PyObject *callback_id_obj = PyLong_FromUnsignedLongLong(tl_pending_callback_id);
+                        PyObject *func_name_obj = PyUnicode_FromStringAndSize(
+                            tl_pending_func_name, tl_pending_func_name_len);
 
+                        if (callback_id_obj == NULL || func_name_obj == NULL) {
+                            Py_XDECREF(callback_id_obj);
+                            Py_XDECREF(func_name_obj);
                             Py_DECREF(exc_args);
-
-                            req->result = enif_make_tuple4(env,
-                                ATOM_SUSPENDED,
-                                callback_id_term,
-                                state_ref,
-                                enif_make_tuple2(env, func_name_term, args_term));
+                            tl_pending_callback = false;
+                            req->result = make_error(env, "build_exc_args_failed");
+                        } else {
+                            PyTuple_SET_ITEM(exc_args, 0, callback_id_obj);
+                            PyTuple_SET_ITEM(exc_args, 1, func_name_obj);
+                            Py_INCREF(tl_pending_args);
+                            PyTuple_SET_ITEM(exc_args, 2, tl_pending_args);
+
+                            suspended_state_t *suspended = create_suspended_state(env, exc_args, req);
+                            if (suspended == NULL) {
+                                Py_DECREF(exc_args);
+                                tl_pending_callback = false;
+                                req->result = make_error(env, "create_suspended_state_failed");
+                            } else {
+                                ERL_NIF_TERM state_ref = enif_make_resource(env, suspended);
+                                enif_release_resource(suspended);
+
+                                ERL_NIF_TERM callback_id_term = enif_make_uint64(env, tl_pending_callback_id);
+
+                                ERL_NIF_TERM func_name_term;
+                                unsigned char *fn_buf = enif_make_new_binary(env, tl_pending_func_name_len, &func_name_term);
+                                memcpy(fn_buf, tl_pending_func_name, tl_pending_func_name_len);
+
+                                ERL_NIF_TERM args_term = py_to_term(env, tl_pending_args);
+
+                                Py_DECREF(exc_args);
+                                tl_pending_callback = false;
+
+                                req->result = enif_make_tuple4(env,
+                                    ATOM_SUSPENDED,
+                                    callback_id_term,
+                                    state_ref,
+                                    enif_make_tuple2(env, func_name_term, args_term));
+                            }
                         }
                     }
                 } else {

c_src/py_nif.c

@@ -83,6 +83,13 @@ __thread ErlNifEnv *tl_callback_env = NULL;
 __thread suspended_state_t *tl_current_suspended = NULL;
 __thread bool tl_allow_suspension = false;
 
+/* Thread-local pending callback state (flag-based detection, not exception-based) */
+__thread bool tl_pending_callback = false;
+__thread uint64_t tl_pending_callback_id = 0;
+__thread char *tl_pending_func_name = NULL;
+__thread size_t tl_pending_func_name_len = 0;
+__thread PyObject *tl_pending_args = NULL;
+
 /* Thread-local timeout state */
 __thread uint64_t tl_timeout_deadline = 0;
 __thread bool tl_timeout_enabled = false;

c_src/py_nif.h

@@ -717,6 +717,13 @@ extern __thread suspended_state_t *tl_current_suspended;
 /** @brief Flag: suspension is allowed in current context */
 extern __thread bool tl_allow_suspension;
 
+/** @brief Flag: pending callback detected (checked before exception type) */
+extern __thread bool tl_pending_callback;
+extern __thread uint64_t tl_pending_callback_id;
+extern __thread char *tl_pending_func_name;
+extern __thread size_t tl_pending_func_name_len;
+extern __thread PyObject *tl_pending_args;
+
 /** @brief Timeout deadline (nanoseconds, monotonic clock) */
 extern __thread uint64_t tl_timeout_deadline;