Fix superseded async references escaping cancellation

param/parameterized.py

@@ -2579,10 +2579,11 @@ def __getattr__(self_, attr: str) -> t.Any:
 
         ns_type = type(self_)
 
-        # The cached parameters are read from _param__private directly rather
-        # than via the _cls_parameters property. If that property raised an
-        # AttributeError we would be called to handle it and recurse, so it is
-        # only invoked as a descriptor, i.e. without attribute lookup on self_.
+        # Read the cached parameters from _param__private rather than from the
+        # _cls_parameters property: an AttributeError raised by that property is
+        # dispatched to this method, so accessing it as an attribute here would
+        # recurse. When the cache is empty the property is invoked directly as a
+        # descriptor instead, which bypasses that dispatch.
         params = cls._param__private.params
         if not params:
             params = _find_descriptor(ns_type, '_cls_parameters').__get__(self_, ns_type)
@@ -2742,7 +2743,10 @@ def _sync_refs(self_, *events):
             if new_val is Skip or new_val is Undefined:
                 continue
             elif is_async:
-                async_executor(partial(self_._async_ref, pname, t.cast("t.Awaitable[t.Any]", new_val)))
+                generation = self_._schedule_async_ref(pname)
+                async_executor(partial(
+                    self_._async_ref, pname, t.cast("t.Awaitable[t.Any]", new_val), generation
+                ))
                 continue
 
             updates[pname] = new_val
@@ -2763,39 +2767,85 @@ def _resolve_ref(self_, pobj: Parameter, value: t.Any):
         except Skip:
             value = Undefined
         if is_async and pobj.name:
-            async_executor(partial(self_._async_ref, pobj.name, t.cast("t.Awaitable[t.Any]", value)))
+            generation = self_._schedule_async_ref(pobj.name)
+            async_executor(partial(
+                self_._async_ref, pobj.name, t.cast("t.Awaitable[t.Any]", value), generation
+            ))
             value = None
         return ref, deps, value, is_async
 
-    async def _async_ref(self_, pname: str, awaitable: t.Awaitable[t.Any]):
+    def _schedule_async_ref(self_, pname: str) -> int:
+        """
+        Record that an asynchronous reference resolution is about to be
+        scheduled and return the generation identifying it.
+
+        The generation is bumped synchronously, before the task is handed to
+        the executor, so that the reference reads as unsettled from the moment
+        it is superseded rather than only once the task starts running.
+        """
+        if self_.self is None:
+            return 0
+        private = self_.self._param__private
+        private.async_ref_scheduled[pname] += 1
+        return private.async_ref_scheduled[pname]
+
+    def _settle_async_ref(self_, pname: str, generation: int):
+        """
+        Record that the resolution identified by ``generation`` produced a
+        value, or gave up on producing one.
+
+        A superseded task settling late must not mark the reference settled for
+        the generation that superseded it, so the generation is recorded rather
+        than cleared. Generators settle on every value they yield, so a
+        reference is unsettled only until its next value arrives, not until the
+        generator is exhausted.
+        """
+        if self_.self is None or not generation:
+            return
+        settled = self_.self._param__private.async_ref_settled
+        settled[pname] = max(settled[pname], generation)
+
+    def _awaiting_ref(self_, pname: str) -> bool:
+        """Whether an asynchronous reference has not yet produced a value."""
+        if self_.self is None:
+            return False
+        private = self_.self._param__private
+        return private.async_ref_scheduled[pname] != private.async_ref_settled[pname]
+
+    async def _async_ref(self_, pname: str, awaitable: t.Awaitable[t.Any], generation: int = 0):
         if self_.self is None:
             return
         if not self_.self._param__private.initialized:
-            async_executor(partial(self_._async_ref, pname, awaitable))
+            async_executor(partial(self_._async_ref, pname, awaitable, generation))
             return
 
         import asyncio
         current_task = asyncio.current_task()
         running_task = self_.self._param__private.async_refs.get(pname)
-        if running_task is None:
+        if running_task is not current_task:
+            if running_task is not None:
+                running_task.cancel()
             self_.self._param__private.async_refs[pname] = current_task
-        elif current_task is not running_task:
-            self_.self._param__private.async_refs[pname].cancel()
         try:
             if isinstance(awaitable, types.AsyncGeneratorType):
                 async for new_obj in awaitable:
                     with _syncing(self_.self, (pname,)):
                         self_.update({pname: new_obj})
+                    self_._settle_async_ref(pname, generation)
             else:
                 with _syncing(self_.self, (pname,)):
                     try:
                         self_.update({pname: await awaitable})
                     except Skip:
                         pass
+                self_._settle_async_ref(pname, generation)
         finally:
-            # Ensure we clean up but only if the task matches the current task
-            if self_.self._param__private.async_refs.get(pname) is current_task:
-                del self_.self._param__private.async_refs[pname]
+            self_._settle_async_ref(pname, generation)
+            # Ensure we clean up but only if the task matches the current task,
+            # i.e. only the resolution that still owns the reference clears it.
+            async_refs = self_.self._param__private.async_refs
+            if pname in async_refs and async_refs[pname] is current_task:
+                del async_refs[pname]
 
     @classmethod
     def _changed(cls, event):
@@ -5638,6 +5688,8 @@ class _InstancePrivate:
         'dynamic_watchers',
         'params',
         'async_refs',
+        'async_ref_scheduled',
+        'async_ref_settled',
         'refs',
         'ref_watchers',
         'syncing',
@@ -5651,6 +5703,8 @@ class _InstancePrivate:
     dynamic_watchers: defaultdict[str, list[Watcher]]
     params: dict[str, Parameter]
     async_refs: dict[str, t.Any]
+    async_ref_scheduled: defaultdict[str, int]
+    async_ref_settled: defaultdict[str, int]
     refs: dict[str, t.Any]
     ref_watchers: list[tuple[tuple[str, ...], Watcher]]
     syncing: set[str]
@@ -5681,6 +5735,8 @@ def __init__(
             }
         self.ref_watchers = []
         self.async_refs = {}
+        self.async_ref_scheduled = defaultdict(int)
+        self.async_ref_settled = defaultdict(int)
         self.parameters_state = parameters_state
         self.dynamic_watchers = defaultdict(list, dynamic_watchers or ())
         self.params = {} if params is None else params

tests/testrefs.py

@@ -304,6 +304,55 @@ async def gen_strings2():
     assert task2 is async_task2
     assert p._param__private.async_refs['string'] is task2
 
+async def test_async_ref_cancelled_on_dependency_change():
+    started, finished = [], []
+
+    class Source(param.Parameterized):
+        x = param.Number(default=0)
+
+    async def slow(i):
+        started.append(i)
+        await asyncio.sleep(0.1)
+        finished.append(i)
+        return str(i)
+
+    source = Source()
+    p = Parameters(string=bind(slow, source.param.x))
+    for i in (1, 2, 3):
+        source.x = i
+        await asyncio.sleep(0.01)
+
+    await wait_for_value(p, 'string', '3', timeout=1)
+    assert started == [0, 1, 2, 3]
+    assert finished == [3]
+
+    await asyncio.sleep(0.05)
+    assert 'string' not in p._param__private.async_refs
+
+async def test_async_generator_ref_cancelled_on_dependency_change():
+    emitted = []
+
+    class Source(param.Parameterized):
+        x = param.Number(default=0)
+
+    async def gen(i):
+        while True:
+            emitted.append(i)
+            yield str(i)
+            await asyncio.sleep(0.01)
+
+    source = Source()
+    p = Parameters(string=bind(gen, source.param.x))
+    await asyncio.sleep(0.05)
+    source.x = 1
+    await asyncio.sleep(0.05)
+    source.x = 2
+    await wait_for_value(p, 'string', '2', timeout=0.5)
+
+    emitted.clear()
+    await asyncio.sleep(0.05)
+    assert set(emitted) == {2}
+
 async def test_generator_ref_cancelled():
     threads = []
     def gen_strings1():