Fix memory leak on short-lived reactive pipelines

Derived rx nodes registered their invalidation callbacks
(_invalidate_obj/_invalidate_current) on their source parameter as strong
bound-method references. Since the source is typically long-lived, this
pinned every derived node — and anything it captured, such as large arrays
— alive indefinitely, with no way to reclaim it.

Wrap the invalidation callbacks in a weakref.WeakMethod (_WeakInvalidator)
and register a finalizer to prune the watcher once the node is collected.
The node now becomes garbage collectable as soon as it is no longer
referenced, and the source's watcher list no longer grows without bound.

This changes no observable public API behavior: while an expression is
referenced it behaves exactly as before, and watch()/_watch() are
untouched. Only a node that was previously unreachable-but-uncollectable
(pinned solely by the source's internal watcher) is now reclaimed.

Assisted-by: Claude:claude-opus-4-8

Author: hoxbro

param/reactive.py

@@ -94,6 +94,7 @@
 import operator
 import typing as t
 import warnings
+import weakref
 
 from collections.abc import (
     AsyncGenerator, Callable, Coroutine, Generator, Iterable, Iterator, Sized
@@ -1359,6 +1360,39 @@ def wrapped_sync(*wargs, **wkwargs):
         setattr(wrapped, name, t.cast('Callable', accessor)(wrapped))
     return wrapped
 
+class _WeakInvalidator:
+    """
+    A weakly-bound invalidation callback for an ``rx`` pipeline node.
+
+    ``rx`` nodes register invalidation callbacks on their *source* parameters.
+    If those callbacks were strong bound-method references the source (which is
+    typically long-lived) would keep every derived node — and anything it
+    captured, such as large arrays — alive indefinitely. Wrapping the bound
+    method in a :class:`weakref.WeakMethod` lets the derived node be garbage
+    collected as soon as nothing else references it; once collected the call
+    becomes a no-op and the now-dead watcher is removed by a finalizer (see
+    ``rx._watch_invalidation``).
+    """
+
+    __slots__ = ('_ref', '__weakref__')
+
+    def __init__(self, method):
+        self._ref = weakref.WeakMethod(method)
+
+    def __call__(self, *events):
+        method = self._ref()
+        if method is not None:
+            return method(*events)
+
+
+def _remove_watcher(owner, watcher):
+    """Unwatch ``watcher`` from ``owner``, ignoring if it is already gone."""
+    try:
+        owner.param.unwatch(watcher)
+    except Exception:
+        pass
+
+
 # When we only support python >= 3.11 we should exchange 'rx' with Self type annotation below.
 # See https://peps.python.org/pep-0673/
 
@@ -1712,9 +1746,21 @@ def _setup_invalidations(self, depth: int = 0):
             for _, params in full_groupby(self._fn_params, lambda x: id(x.owner)):
                 fps = [p.name for p in params if p in self._root._fn_params]
                 if fps:
-                    params[0].owner.param._watch(self._invalidate_obj, fps, precedence=-1)
+                    self._watch_invalidation(params[0].owner, self._invalidate_obj, fps)
         for _, params in full_groupby(self._internal_params, lambda x: id(x.owner)):
-            params[0].owner.param._watch(self._invalidate_current, [p.name for p in params], precedence=-1)
+            self._watch_invalidation(params[0].owner, self._invalidate_current, [p.name for p in params])
+
+    def _watch_invalidation(self, owner, method, names):
+        """
+        Register a *weak* invalidation watcher on a source parameter.
+
+        The callback only holds a weak reference to this node, so a long-lived
+        source does not pin the (potentially short-lived) derived node alive.
+        A finalizer removes the watcher automatically once this node is garbage
+        collected, keeping the source's watcher list from growing without bound.
+        """
+        watcher = owner.param._watch(_WeakInvalidator(method), names, precedence=-1)
+        weakref.finalize(self, _remove_watcher, owner, watcher)
 
     def _invalidate_current(self, *events):
         if all(event.obj is self._trigger for event in events):

tests/testreactive.py

@@ -1,8 +1,10 @@
 import asyncio
+import gc
 import math
 import operator
 import re
 import time
+import weakref
 
 import param
 import pytest
@@ -1029,3 +1031,28 @@ async def expensive_compute(a):
     assert y_rx.rx.value == 4
 
     assert call_count == 2
+
+
+def test_reactive_derived_node_is_garbage_collected():
+    def watcher_count(source):
+        watchers = source._internal_params[0].owner._param__private.watchers
+        return sum(len(lst) for what in watchers.values() for lst in what.values())
+
+    a = param.rx(1)
+    baseline = watcher_count(a)
+    assert baseline == 2
+
+    b = a + 1
+    assert b.rx.value == 2
+    assert watcher_count(a) > baseline
+
+    ref = weakref.ref(b)
+    del b
+    gc.collect()
+    assert ref() is None
+    assert watcher_count(a) == baseline
+
+    c = a + 10
+    a.rx.value = 5
+    assert c.rx.value == 15
+    assert watcher_count(a) > baseline