Add synchronous blocking channel receive (#23)

* Add synchronous blocking channel receive

Implement blocking receive for channels that suspends Python while
waiting for data and releases the dirty scheduler worker.

- Add sync_waiter_pid and has_sync_waiter fields to py_channel_t
- Add channel_register_sync_waiter NIF to register calling process
- Modify channel_send to notify sync waiter via Erlang message
- Modify channel_close to notify sync waiter of channel closure
- Implement blocking handle_receive using Erlang receive to wait
- Add tests for immediate, delayed, and closed channel cases

* Add async receive e2e test to verify asyncio integration

* Add subinterpreter mode sync receive test

Test verifies blocking channel receive works with Python 3.12+
subinterpreter contexts. Skips gracefully on older Python versions.

* Fix subinterp_sync_receive_wait_test to be proper e2e test

- Use py_context:start_link/2 with unique integer ID
- Use py_context:stop/1 instead of gen_server:stop
- Test immediate receive, blocking receive with delayed send,
  try_receive on empty, and closed channel detection

* Fix race condition and cleanup in sync waiter registration

- Check if data is available when registering sync waiter to handle race
  between try_receive returning empty and register_sync_waiter being called
- Return 'has_data' atom when data arrived in the window, caller retries
- Notify sync waiter in channel destructor when channel is GC'd
- Do not notify async waiter in destructor to avoid use-after-free when
  event loop is destroyed concurrently
- Update test to consume data before re-registering waiter

* Remove unused ChannelBuffer Python type

The ChannelBuffer type was defined but never used. Removing dead code.

* Fix channel waiter race conditions and lost wakeups

- Reject duplicate/mixed waiters: both async and sync waiter registration
  now return {error, waiter_exists} if any waiter already exists
- Fix lost wakeups: event_loop_add_pending now returns bool; waiter state
  is only cleared after successful dispatch
- Add null checks for enif_alloc_env in sync waiter notifications
- Add tests for mixed waiter rejection scenarios

* Remove deprecated ASGI/WSGI NIF tests, keep deprecation checks

The asgi_run and wsgi_run NIF functions are deprecated. Removed tests
that call these functions, keeping only the deprecation attribute tests.

* Fix use-after-free in tl_pending_args across subinterpreters

Clear tl_pending_args to NULL whenever tl_pending_callback is set to
false. Previously, the thread-local pointer was left dangling after
callback completion. When a dirty scheduler thread later handled a
different subinterpreter's code, Py_XDECREF on the stale pointer
would attempt to free memory from the wrong allocator.

Author: benoitc

c_src/py_callback.c

@@ -563,6 +563,7 @@ static PyObject *build_pending_callback_exc_args(void) {
     PyObject *exc_args = PyTuple_New(3);
     if (exc_args == NULL) {
         tl_pending_callback = false;
+        Py_CLEAR(tl_pending_args);
         return NULL;
     }
 
@@ -575,6 +576,7 @@ static PyObject *build_pending_callback_exc_args(void) {
         Py_XDECREF(func_name_obj);
         Py_DECREF(exc_args);
         tl_pending_callback = false;
+        Py_CLEAR(tl_pending_args);
         return NULL;
     }
 
@@ -610,6 +612,7 @@ static ERL_NIF_TERM build_suspended_result(ErlNifEnv *env, suspended_state_t *su
     ERL_NIF_TERM args_term = py_to_term(env, tl_pending_args);
 
     tl_pending_callback = false;
+    Py_CLEAR(tl_pending_args);
 
     return enif_make_tuple4(env,
         ATOM_SUSPENDED,
@@ -811,6 +814,7 @@ static ERL_NIF_TERM build_suspended_context_result(ErlNifEnv *env, suspended_con
     ERL_NIF_TERM args_term = py_to_term(env, tl_pending_args);
 
     tl_pending_callback = false;
+    Py_CLEAR(tl_pending_args);
 
     return enif_make_tuple4(env,
         ATOM_SUSPENDED,
@@ -1290,6 +1294,19 @@ PyTypeObject ErlangPidType = {
 static PyObject *erlang_call_impl(PyObject *self, PyObject *args) {
     (void)self;
 
+    /*
+     * Invariant check: pending callback TLS must be clear when entering.
+     * If any state is still set, it's leaked from a prior context that didn't
+     * properly clean up - fail loudly rather than risk cross-interpreter corruption.
+     */
+    if (tl_pending_callback || tl_pending_args != NULL ||
+        tl_pending_func_name != NULL || tl_pending_callback_id != 0) {
+        PyErr_SetString(PyExc_RuntimeError,
+            "erlang.call: stale pending callback TLS detected - "
+            "prior context did not clean up properly");
+        return NULL;
+    }
+
     /*
      * Check if we have a callback handler available.
      * Priority:
@@ -1553,6 +1570,7 @@ static PyObject *erlang_call_impl(PyObject *self, PyObject *args) {
     tl_pending_func_name = enif_alloc(func_name_len + 1);
     if (tl_pending_func_name == NULL) {
         tl_pending_callback = false;
+        Py_CLEAR(tl_pending_args);
         Py_DECREF(call_args);
         PyErr_SetString(PyExc_MemoryError, "Failed to allocate function name");
         return NULL;
@@ -1561,9 +1579,12 @@ static PyObject *erlang_call_impl(PyObject *self, PyObject *args) {
     tl_pending_func_name[func_name_len] = '\0';
     tl_pending_func_name_len = func_name_len;
 
-    /* Store args (take ownership) */
-    Py_XDECREF(tl_pending_args);
-    tl_pending_args = call_args;  /* Takes ownership, don't decref */
+    /* Store args (take ownership)
+     * Use Py_XSETREF for swap-first pattern: sets tl_pending_args to new value
+     * BEFORE decref'ing old value. This prevents re-entrancy issues if the old
+     * object's finalizer triggers another erlang.call() during decref.
+     */
+    Py_XSETREF(tl_pending_args, call_args);
 
     /* Raise exception to abort Python execution */
     PyErr_SetString(SuspensionRequiredException, "callback pending");
@@ -2649,6 +2670,7 @@ static ERL_NIF_TERM nif_resume_callback_dirty(ErlNifEnv *env, int argc, const ER
                     Py_DECREF(exc_args);
                     if (new_suspended == NULL) {
                         tl_pending_callback = false;
+                        Py_CLEAR(tl_pending_args);
                         result = make_error(env, "create_nested_suspended_state_failed");
                     } else {
                         result = build_suspended_result(env, new_suspended);
@@ -2717,6 +2739,7 @@ static ERL_NIF_TERM nif_resume_callback_dirty(ErlNifEnv *env, int argc, const ER
                         Py_DECREF(exc_args);
                         if (new_suspended == NULL) {
                             tl_pending_callback = false;
+                            Py_CLEAR(tl_pending_args);
                             result = make_error(env, "create_nested_suspended_state_failed");
                         } else {
                             result = build_suspended_result(env, new_suspended);