Fix flag-based callback detection in replay path

The previous commit added flag-based detection in process_request but
missed the replay path in nif_resume_callback_dirty. This caused
SuspensionRequired exceptions to still leak when ASGI/WSGI middleware
caught and re-raised exceptions during replay.

Changes:
- Update nif_resume_callback_dirty to check tl_pending_callback flag
  first, for both CALL and EVAL replay paths
- Remove unused is_suspension_exception() and get_suspension_args()
- Add test_callback_with_try_except test with py_test_middleware.py
  module that tests try/except, BaseException, nested try/except,
  try/finally, and multi-layer middleware patterns

Author: benoitc

c_src/py_callback.c

@@ -87,46 +87,6 @@
  * Suspended state management
  * ============================================================================ */
 
-/**
- * Check if a SuspensionRequired exception is pending.
- * Returns true if the exception is set and matches SuspensionRequiredException.
- */
-static bool is_suspension_exception(void) {
-    if (!PyErr_Occurred()) {
-        return false;
-    }
-    return PyErr_ExceptionMatches(SuspensionRequiredException);
-}
-
-/**
- * Extract suspension info from the pending SuspensionRequired exception.
- * Returns the exception args tuple (callback_id, func_name, args) or NULL.
- * Clears the exception if successful.
- */
-static PyObject *get_suspension_args(void) {
-    PyObject *exc_type, *exc_value, *exc_tb;
-    PyErr_Fetch(&exc_type, &exc_value, &exc_tb);
-
-    if (exc_value == NULL) {
-        Py_XDECREF(exc_type);
-        Py_XDECREF(exc_tb);
-        return NULL;
-    }
-
-    /* Get the args from the exception - it's a tuple (callback_id, func_name, args) */
-    PyObject *args = PyObject_GetAttrString(exc_value, "args");
-    Py_DECREF(exc_type);
-    Py_DECREF(exc_value);
-    Py_XDECREF(exc_tb);
-
-    if (args == NULL || !PyTuple_Check(args) || PyTuple_Size(args) != 3) {
-        Py_XDECREF(args);
-        return NULL;
-    }
-
-    return args;  /* Caller owns this reference */
-}
-
 /**
  * Create a suspended state resource from exception args.
  * Args tuple format: (callback_id, func_name, args)
@@ -1193,23 +1153,46 @@ static ERL_NIF_TERM nif_resume_callback_dirty(ErlNifEnv *env, int argc, const ER
         Py_XDECREF(kwargs);
 
         if (py_result == NULL) {
-            if (is_suspension_exception()) {
+            if (tl_pending_callback) {
                 /*
-                 * Another suspension during replay - Python made a second erlang.call().
-                 * Create a new suspended state and return {suspended, ...} so Erlang
-                 * can handle this callback and resume again.
+                 * Flag-based callback detection during replay.
+                 * 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) {
-                    result = make_error(env, "get_suspension_args_failed");
+                    tl_pending_callback = false;
+                    result = make_error(env, "alloc_exc_args_failed");
                 } else {
-                    suspended_state_t *new_suspended = create_suspended_state_from_existing(env, exc_args, state);
-                    if (new_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);
-                        result = make_error(env, "create_nested_suspended_state_failed");
+                        tl_pending_callback = false;
+                        result = make_error(env, "build_exc_args_failed");
                     } else {
-                        result = make_suspended_term(env, new_suspended, exc_args);
-                        Py_DECREF(exc_args);
+                        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 *new_suspended = create_suspended_state_from_existing(env, exc_args, state);
+                        if (new_suspended == NULL) {
+                            Py_DECREF(exc_args);
+                            tl_pending_callback = false;
+                            result = make_error(env, "create_nested_suspended_state_failed");
+                        } else {
+                            result = make_suspended_term(env, new_suspended, exc_args);
+                            Py_DECREF(exc_args);
+                            tl_pending_callback = false;
+                        }
                     }
                 }
             } else {
@@ -1258,23 +1241,46 @@ static ERL_NIF_TERM nif_resume_callback_dirty(ErlNifEnv *env, int argc, const ER
             Py_DECREF(compiled);
 
             if (py_result == NULL) {
-                if (is_suspension_exception()) {
+                if (tl_pending_callback) {
                     /*
-                     * Another suspension during replay - Python made a second erlang.call().
-                     * Create a new suspended state and return {suspended, ...} so Erlang
-                     * can handle this callback and resume again.
+                     * Flag-based callback detection during eval replay.
+                     * 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) {
-                        result = make_error(env, "get_suspension_args_failed");
+                        tl_pending_callback = false;
+                        result = make_error(env, "alloc_exc_args_failed");
                     } else {
-                        suspended_state_t *new_suspended = create_suspended_state_from_existing(env, exc_args, state);
-                        if (new_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);
-                            result = make_error(env, "create_nested_suspended_state_failed");
+                            tl_pending_callback = false;
+                            result = make_error(env, "build_exc_args_failed");
                         } else {
-                            result = make_suspended_term(env, new_suspended, exc_args);
-                            Py_DECREF(exc_args);
+                            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 *new_suspended = create_suspended_state_from_existing(env, exc_args, state);
+                            if (new_suspended == NULL) {
+                                Py_DECREF(exc_args);
+                                tl_pending_callback = false;
+                                result = make_error(env, "create_nested_suspended_state_failed");
+                            } else {
+                                result = make_suspended_term(env, new_suspended, exc_args);
+                                Py_DECREF(exc_args);
+                                tl_pending_callback = false;
+                            }
                         }
                     }
                 } else {

c_src/py_nif.h

@@ -1107,23 +1107,6 @@ static PyObject *erlang_module_getattr(PyObject *module, PyObject *name);
  */
 static void *async_event_loop_thread(void *arg);
 
-/**
- * @brief Check if SuspensionRequired exception is pending
- *
- * @return true if exception matches SuspensionRequiredException
- */
-static bool is_suspension_exception(void);
-
-/**
- * @brief Extract args from pending SuspensionRequired exception
- *
- * Gets (callback_id, func_name, args) tuple from exception and
- * clears the exception state.
- *
- * @return New reference to args tuple, or NULL on error
- */
-static PyObject *get_suspension_args(void);
-
 /**
  * @brief Create suspended state for callback handling
  *

test/py_reentrant_SUITE.erl

@@ -21,7 +21,8 @@
     test_concurrent_reentrant/1,
     test_callback_with_complex_types/1,
     test_multiple_sequential_callbacks/1,
-    test_call_from_non_worker_thread/1
+    test_call_from_non_worker_thread/1,
+    test_callback_with_try_except/1
 ]).
 
 all() ->
@@ -32,7 +33,8 @@ all() ->
         test_concurrent_reentrant,
         test_callback_with_complex_types,
         test_multiple_sequential_callbacks,
-        test_call_from_non_worker_thread
+        test_call_from_non_worker_thread,
+        test_callback_with_try_except
     ].
 
 init_per_suite(Config) ->
@@ -259,3 +261,50 @@ test_call_from_non_worker_thread(_Config) ->
     %% Cleanup
     py:unregister_function(simple_add),
     ok.
+
+%% @doc Test that erlang.call() works even when wrapped in try/except blocks.
+%% This simulates ASGI/WSGI middleware that catches all exceptions.
+%% The flag-based detection should work even when the SuspensionRequired
+%% exception is caught and re-raised by Python code.
+%%
+%% Uses py:call on a test module with try/except blocks to properly test
+%% the suspension mechanism through middleware-like exception handling.
+test_callback_with_try_except(_Config) ->
+    %% Register a simple Erlang function
+    py:register_function(get_value, fun([Key]) ->
+        case Key of
+            <<"a">> -> 1;
+            <<"b">> -> 2;
+            <<"c">> -> 3;
+            _ -> 0
+        end
+    end),
+
+    %% Add test directory to Python path so we can import the test module
+    TestDir = code:lib_dir(erlang_python, test),
+    ok = py:exec(iolist_to_binary(io_lib:format(
+        "import sys; sys.path.insert(0, '~s')", [TestDir]))),
+
+    %% Test 1: Try/except that catches Exception and re-raises
+    {ok, Result1} = py:call(py_test_middleware, call_with_try_except, [<<"a">>]),
+    1 = Result1,
+
+    %% Test 2: Try/except that catches BaseException (catches everything)
+    {ok, Result2} = py:call(py_test_middleware, call_with_base_exception, [<<"b">>]),
+    2 = Result2,
+
+    %% Test 3: Nested try/except blocks
+    {ok, Result3} = py:call(py_test_middleware, call_with_nested_try, [<<"c">>]),
+    3 = Result3,
+
+    %% Test 4: Try/finally pattern
+    {ok, Result4} = py:call(py_test_middleware, call_with_finally, [<<"a">>]),
+    1 = Result4,
+
+    %% Test 5: Multiple middleware layers
+    {ok, Result5} = py:call(py_test_middleware, call_through_layers, [<<"b">>]),
+    2 = Result5,
+
+    %% Cleanup
+    py:unregister_function(get_value),
+    ok.

test/py_test_middleware.py

@@ -0,0 +1,63 @@
+"""Test module for middleware-style exception handling.
+
+This module tests that erlang.call() works correctly even when wrapped
+in try/except blocks that catch and re-raise exceptions, simulating
+ASGI/WSGI middleware behavior.
+"""
+
+import erlang
+
+
+def call_with_try_except(key):
+    """Wraps erlang.call in try/except that catches and re-raises."""
+    try:
+        return erlang.call('get_value', key)
+    except Exception as e:
+        # This simulates middleware logging
+        raise
+
+
+def call_with_base_exception(key):
+    """Wraps erlang.call in try/except that catches BaseException."""
+    try:
+        return erlang.call('get_value', key)
+    except BaseException as e:
+        # Catches everything including SuspensionRequired
+        raise
+
+
+def call_with_nested_try(key):
+    """Multiple nested try/except blocks."""
+    try:
+        try:
+            return erlang.call('get_value', key)
+        except Exception:
+            raise
+    except Exception:
+        raise
+
+
+def call_with_finally(key):
+    """Try/finally pattern common in cleanup code."""
+    cleanup_ran = False
+    try:
+        return erlang.call('get_value', key)
+    finally:
+        cleanup_ran = True
+
+
+def call_through_layers(key):
+    """Simulates calling through multiple middleware layers."""
+    def inner():
+        try:
+            return erlang.call('get_value', key)
+        except Exception:
+            raise
+
+    def outer():
+        try:
+            return inner()
+        except Exception:
+            raise
+
+    return outer()