Fix event loop isolation, atom safety, and Python 3.14 compat

benoitc · benoitc · commit 1b87ea2adf12 · 2026-03-16T09:43:50.000+01:00
- Add _has_loop_ref() to prevent concurrent loops while allowing
  sequential replacement (checks is_running() not just exists)
- Add _clear_loop_ref() called on loop close for proper cleanup
- Add global_loop_capsule_destructor to fix resource leak
- Rename atom() to _atom() in C, add Python wrapper with cache
  and configurable limit (ERLANG_PYTHON_MAX_ATOMS, default 10000)
- Use enif_make_existing_atom() first to avoid duplicate atoms
- Fix venv .pth file processing for Python 3.14 subinterpreters
  by embedding site-packages path directly in exec code
diff --git a/c_src/py_callback.c b/c_src/py_callback.c
@@ -2862,10 +2862,11 @@ static PyMethodDef ErlangModuleMethods[] = {
      "Call a registered Erlang function.\n\n"
      "Usage: erlang.call('func_name', arg1, arg2, ...)\n"
      "Returns: The result from the Erlang function."},
-    {"atom", erlang_atom_impl, METH_VARARGS,
-     "Create an Erlang atom.\n\n"
-     "Usage: erlang.atom('name')\n"
-     "Returns: An ErlangAtom object that converts to an Erlang atom."},
+    {"_atom", erlang_atom_impl, METH_VARARGS,
+     "Internal: Create an Erlang atom.\n\n"
+     "Usage: erlang._atom('name')\n"
+     "Returns: An ErlangAtom object that converts to an Erlang atom.\n"
+     "NOTE: Use erlang.atom() wrapper instead for safety limits."},
     {"send", erlang_send_impl, METH_VARARGS,
      "Send a message to an Erlang process (fire-and-forget).\n\n"
      "Usage: erlang.send(pid, term)\n"
diff --git a/c_src/py_convert.c b/c_src/py_convert.c
@@ -353,6 +353,12 @@ ERL_NIF_TERM py_to_term(ErlNifEnv *env, PyObject *obj) {
     /* Handle ErlangAtom → Erlang atom */
     if (Py_IS_TYPE(obj, &ErlangAtomType)) {
         ErlangAtomObject *atom_obj = (ErlangAtomObject *)obj;
+        ERL_NIF_TERM atom_term;
+        /* Try existing atom first (no new allocation) */
+        if (enif_make_existing_atom(env, atom_obj->name, &atom_term, ERL_NIF_LATIN1)) {
+            return atom_term;
+        }
+        /* Atom doesn't exist yet, create it */
         return enif_make_atom(env, atom_obj->name);
     }
 
diff --git a/c_src/py_event_loop.c b/c_src/py_event_loop.c
@@ -6176,6 +6176,20 @@ static void loop_capsule_destructor(PyObject *capsule) {
     }
 }
 
+/**
+ * Destructor for global loop capsules.
+ * Only releases reference - does NOT signal shutdown since the global
+ * loop is shared and managed by Erlang, not Python.
+ */
+static void global_loop_capsule_destructor(PyObject *capsule) {
+    erlang_event_loop_t *loop = (erlang_event_loop_t *)PyCapsule_GetPointer(
+        capsule, LOOP_CAPSULE_NAME);
+    if (loop != NULL) {
+        /* Only release the reference, don't shutdown */
+        enif_release_resource(loop);
+    }
+}
+
 /* Python function: _loop_new() -> capsule */
 static PyObject *py_loop_new(PyObject *self, PyObject *args) {
     (void)self;
@@ -6384,6 +6398,36 @@ static PyObject *py_set_global_loop_ref(PyObject *self, PyObject *args) {
     Py_RETURN_NONE;
 }
 
+/**
+ * Python function: _clear_loop_ref(capsule)
+ *
+ * Clear the Python loop reference from an event loop capsule.
+ * Should be called when the Python loop is closed to allow
+ * creating a new loop later.
+ */
+static PyObject *py_clear_loop_ref(PyObject *self, PyObject *args) {
+    (void)self;
+    PyObject *capsule;
+
+    if (!PyArg_ParseTuple(args, "O", &capsule)) {
+        return NULL;
+    }
+
+    erlang_event_loop_t *loop = loop_from_capsule(capsule);
+    if (loop == NULL) {
+        return NULL;
+    }
+
+    /* Clear the Python loop reference */
+    if (loop->py_loop != NULL) {
+        Py_DECREF(loop->py_loop);
+        loop->py_loop = NULL;
+    }
+    loop->py_loop_valid = false;
+
+    Py_RETURN_NONE;
+}
+
 /* Python function: _get_global_loop_capsule() -> capsule
  *
  * Returns a capsule for the global interpreter event loop.
@@ -6405,7 +6449,67 @@ static PyObject *py_get_global_loop_capsule(PyObject *self, PyObject *args) {
     /* Keep the resource alive while capsule exists */
     enif_keep_resource(loop);
 
-    return PyCapsule_New(loop, LOOP_CAPSULE_NAME, NULL);
+    return PyCapsule_New(loop, LOOP_CAPSULE_NAME, global_loop_capsule_destructor);
+}
+
+/**
+ * Python function: _has_loop_ref(capsule) -> bool
+ *
+ * Check if a loop capsule has an ACTIVE Python loop reference.
+ * Returns True only if there's a valid loop that is currently RUNNING.
+ * This prevents multiple concurrent loops while allowing sequential
+ * loop replacement (e.g., between test cases).
+ *
+ * The key insight is that the event confusion bug occurs when multiple
+ * loops are running simultaneously. A non-running loop (even if not
+ * explicitly closed) can be safely replaced.
+ */
+static PyObject *py_has_loop_ref(PyObject *self, PyObject *args) {
+    (void)self;
+    PyObject *capsule;
+
+    if (!PyArg_ParseTuple(args, "O", &capsule)) {
+        return NULL;
+    }
+
+    erlang_event_loop_t *loop = loop_from_capsule(capsule);
+    if (loop == NULL) {
+        return NULL;
+    }
+
+    if (loop->py_loop_valid && loop->py_loop != NULL) {
+        /* Check if the existing loop is running - only block if running */
+        PyObject *is_running = PyObject_CallMethod(loop->py_loop, "is_running", NULL);
+        if (is_running != NULL) {
+            int running = PyObject_IsTrue(is_running);
+            Py_DECREF(is_running);
+            if (running) {
+                /* Loop is still running - prevent concurrent loop creation */
+                Py_RETURN_TRUE;
+            }
+        } else {
+            /* Error calling is_running - clear error and check is_closed as fallback */
+            PyErr_Clear();
+        }
+
+        /* Loop exists but is not running - check if closed for cleanup */
+        PyObject *is_closed = PyObject_CallMethod(loop->py_loop, "is_closed", NULL);
+        if (is_closed != NULL) {
+            int closed = PyObject_IsTrue(is_closed);
+            Py_DECREF(is_closed);
+            if (closed) {
+                /* Loop is closed, clean up reference */
+                Py_DECREF(loop->py_loop);
+                loop->py_loop = NULL;
+                loop->py_loop_valid = false;
+            }
+        } else {
+            PyErr_Clear();
+        }
+        /* Not running, allow replacement */
+        Py_RETURN_FALSE;
+    }
+    Py_RETURN_FALSE;
 }
 
 /* Python function: _run_once_native_for(capsule, timeout_ms) -> [(callback_id, event_type), ...] */
@@ -7031,6 +7135,8 @@ static PyMethodDef PyEventLoopMethods[] = {
     /* Handle-based API (takes explicit loop capsule) */
     {"_loop_new", py_loop_new, METH_NOARGS, "Create a new event loop, returns capsule"},
     {"_get_global_loop_capsule", py_get_global_loop_capsule, METH_NOARGS, "Get capsule for global event loop"},
+    {"_has_loop_ref", py_has_loop_ref, METH_VARARGS, "Check if loop capsule has Python loop reference"},
+    {"_clear_loop_ref", py_clear_loop_ref, METH_VARARGS, "Clear Python loop reference from C struct"},
     {"_loop_destroy", py_loop_destroy, METH_VARARGS, "Destroy an event loop"},
     {"_set_loop_ref", py_set_loop_ref, METH_VARARGS, "Store Python loop reference in C struct"},
     {"_set_global_loop_ref", py_set_global_loop_ref, METH_VARARGS, "Store Python loop reference in global loop"},
diff --git a/priv/_erlang_impl/__init__.py b/priv/_erlang_impl/__init__.py
@@ -44,6 +44,7 @@
     loop.run_until_complete(main())
 """
 
+import os
 import sys
 import asyncio
 import time
@@ -82,12 +83,60 @@
     'Channel',
     'reply',
     'ChannelClosed',
+    'atom',
 ]
 
+# Atom caching with configurable limit to prevent BEAM atom table exhaustion.
+# The BEAM VM has a hard limit (~1M atoms) and crashes when exceeded.
+# This provides a Python-level safety valve well under that limit.
+_MAX_USER_ATOMS = int(os.environ.get('ERLANG_PYTHON_MAX_ATOMS', '10000'))
+_atom_cache = {}
+
 # Re-export for uvloop API compatibility
 EventLoopPolicy = ErlangEventLoopPolicy
 
 
+def atom(name):
+    """Create an Erlang atom with safety limit.
+
+    Atoms in Erlang are permanent and the BEAM VM has a hard limit
+    (~1M atoms). This function provides a Python-level cache with
+    a configurable limit to prevent atom table exhaustion from
+    untrusted Python code.
+
+    Args:
+        name: The atom name as a string.
+
+    Returns:
+        An ErlangAtom object that converts to an Erlang atom.
+
+    Raises:
+        RuntimeError: If the atom limit is reached.
+
+    The limit can be configured via the ERLANG_PYTHON_MAX_ATOMS
+    environment variable (default: 10000).
+
+    Example:
+        >>> import erlang
+        >>> ok = erlang.atom('ok')
+        >>> error = erlang.atom('error')
+    """
+    if name in _atom_cache:
+        return _atom_cache[name]
+
+    if len(_atom_cache) >= _MAX_USER_ATOMS:
+        raise RuntimeError(
+            f"Atom limit ({_MAX_USER_ATOMS}) reached. "
+            "Set ERLANG_PYTHON_MAX_ATOMS env var to increase."
+        )
+
+    # Import erlang module to access internal _atom function
+    import erlang as _erlang
+    result = _erlang._atom(name)
+    _atom_cache[name] = result
+    return result
+
+
 def get_event_loop_policy() -> ErlangEventLoopPolicy:
     """Get an Erlang event loop policy instance.
 
diff --git a/priv/_erlang_impl/_loop.py b/priv/_erlang_impl/_loop.py
@@ -125,9 +125,20 @@ def __init__(self):
             try:
                 self._loop_capsule = self._pel._get_global_loop_capsule()
                 self._uses_global_capsule = True
-            except RuntimeError:
+                # Check if another loop already owns this capsule.
+                # Only one ErlangEventLoop per interpreter is supported.
+                if hasattr(self._pel, '_has_loop_ref') and self._pel._has_loop_ref(self._loop_capsule):
+                    raise RuntimeError(
+                        "An ErlangEventLoop already exists for this interpreter. "
+                        "Only one loop per interpreter is supported."
+                    )
+            except RuntimeError as e:
+                # Re-raise our "already exists" error
+                if "already exists" in str(e):
+                    raise
                 # Fall back to creating a new loop if global not available
                 self._loop_capsule = self._pel._loop_new()
+                self._uses_global_capsule = False
         else:
             self._loop_capsule = self._pel._loop_new()
 
@@ -318,6 +329,15 @@ def close(self):
             self._default_executor.shutdown(wait=True)
             self._default_executor = None
 
+        # Clear loop ref to allow creating a new loop later.
+        # This is important for the global capsule case where the capsule
+        # persists but a new Python loop may be created.
+        if self._loop_capsule is not None and hasattr(self._pel, '_clear_loop_ref'):
+            try:
+                self._pel._clear_loop_ref(self._loop_capsule)
+            except Exception:
+                pass
+
         # Destroy loop capsule (but not if using shared global capsule)
         if not self._uses_global_capsule:
             try:
@@ -1306,15 +1326,16 @@ async def _run_and_send(coro, caller_pid, ref):
         (async_result, ref, (ok, result)) - on success
         (async_result, ref, (error, error_str)) - on failure
 
-    Note: Uses erlang.atom() to create atoms for message keys, since Python
+    Note: Uses cached atom() to create atoms for message keys, since Python
     strings become Erlang binaries but the await function expects atoms.
     """
     import erlang
+    from . import atom  # Use cached version from _erlang_impl
 
     # Create atoms for message keys (strings become binaries, await expects atoms)
-    async_result = erlang.atom('async_result')
-    ok = erlang.atom('ok')
-    error = erlang.atom('error')
+    async_result = atom('async_result')
+    ok = atom('ok')
+    error = atom('error')
 
     try:
         result = await coro
diff --git a/src/py.erl b/src/py.erl
@@ -1040,18 +1040,32 @@ activate_venv_with_site_packages(VenvBin, SitePackages) ->
             {ok, _} = eval(<<"setattr(__import__('sys'), '_active_venv', vp)">>, #{vp => VenvBin}),
             {ok, _} = eval(<<"setattr(__import__('sys'), '_venv_site_packages', sp)">>, #{sp => SitePackages}),
             %% Add site-packages and process .pth files (editable installs)
-            ok = exec(<<"import site as _site, sys as _sys\n"
-                         "_b = frozenset(_sys.path)\n"
-                         "_site.addsitedir(_sys._venv_site_packages)\n"
-                         "_sys.path[:] = [p for p in _sys.path if p not in _b] + [p for p in _sys.path if p in _b]\n"
-                         "del _site, _sys, _b\n">>),
+            %% Note: We embed the site-packages path directly since exec doesn't support
+            %% variables and sys attributes may not persist across calls in subinterpreters
+            SitePackagesStr = binary_to_list(SitePackages),
+            ExecCode = iolist_to_binary([
+                <<"import site as _site, sys as _sys\n">>,
+                <<"_sp = '">>, escape_python_string(SitePackagesStr), <<"'\n">>,
+                <<"_b = frozenset(_sys.path)\n">>,
+                <<"_site.addsitedir(_sp)\n">>,
+                <<"_sys.path[:] = [p for p in _sys.path if p not in _b] + [p for p in _sys.path if p in _b]\n">>,
+                <<"del _site, _sys, _b, _sp\n">>
+            ]),
+            ok = exec(ExecCode),
             ok;
         {ok, false} ->
             {error, {invalid_venv, SitePackages}};
         Error ->
             Error
     end.
 
+%% @private Escape a string for embedding in Python code
+escape_python_string(Str) ->
+    lists:flatmap(fun($') -> "\\'";
+                     ($\\) -> "\\\\";
+                     (C) -> [C]
+                  end, Str).
+
 %% @doc Deactivate the current virtual environment.
 %% Restores sys.path to its original state.
 -spec deactivate_venv() -> ok | {error, term()}.
