Stop rx from propagating a stale value while an async node is awaiting

[philippjfr]

Two defects in the same handful of lines of rx._resolve_async / rx._resolve, both caused by
an async node not distinguishing "the result I hold is current" from "the result I hold was
computed from inputs that have since changed". One leaks a stale value to consumers, the other
wastes computation. They share the generation machinery, so they are fixed together.

Problem 1: a stale value is propagated while an async node is awaiting

When a reactive expression contains an async operation, every consumer downstream of it is
notified with the previous value the moment an input changes, before the new result has
resolved. On the first change that previous value is Undefined, so consumers are handed a
sentinel; on later changes they are handed a value computed from inputs that have already been
superseded.

async def mul_slowly(value):
    await asyncio.sleep(0.02)
    return value * 2

i = rx(1)
doubled = i.rx.pipe(mul_slowly)

items = []
doubled.rx.watch(items.append)
await async_wait_until(lambda: items == [2])

i.rx.value = 2
assert items == [2]   # fails on main: [2, 2]

The watcher fires twice for one input change: once immediately with the stale 2, then again
with 4 when the await completes. The same happens for allow_refs parameters fed from the
expression, and for bind-style consumers, because all three go through the same reference
machinery.

This is not cosmetic. A consumer that renders or persists what it is handed writes the wrong
value and then corrects itself, and a downstream operation is executed on the stale value, so
every node in a chain runs twice per change. With more than one async node in a pipeline the
redundant work multiplies.

Problem 2: superseded async tasks still run to completion

When an async node's inputs change several times without the event loop getting a chance to run
in between, every change spawns a task and every one of those bodies runs to completion
concurrently. Only the last result is used, so the rest is pure waste. For a body that hits a
database or an HTTP API, that is N redundant round trips per burst.

async def body(v):
    started.append(v)
    await asyncio.sleep(0.05)
    finished.append(v)
    return v * 2

expr = rx(1).rx.pipe(body)
...
for v in (2, 3, 4, 5):
    i.rx.value = v          # no await between assignments

On main:

started=[1, 2, 3, 4, 5]  finished=[2, 3, 4, 5]  notified=[10]

Four bodies ran to completion for one burst. Inserting await asyncio.sleep(0) between the
assignments gives finished=[5], which is the intended behaviour, so the defect only shows when
updates arrive within a single synchronous block. That is the common case in practice: a callback
that sets several parameters, a param.update batch, or a widget that writes value and range
together. Note the notification side was already correct here; this is wasted computation, not
wrong output.

Root causes

Problem 1. rx._resolve() catches Skip and returns self._current_, the last resolved
value. That is the documented behaviour for a user-raised Skip (.rx.value deliberately keeps
reporting the last value; test_reactive_skip_value pins it), but the same code path is what an
async node takes while its result is in flight, and _rx_transform turned the resolved value
straight into a reference:

return bind(lambda *_: obj.rx.value, *obj._params)

The watchers behind that bind are param watchers on the expression's upstream parameters, not
on the async result. They fire synchronously when an input changes, pull obj.rx.value, and get
the stale value back with nothing distinguishing it from a real one. Skip was raised internally
and then discarded.

Problem 2. _lazy_resolve cancels the previous task before queueing a new one:

previous_task = self._current_task
if previous_task is not None and not previous_task.done():
    previous_task.cancel()
async_executor(partial(self._resolve_async, obj, generation))

But _current_task was only assigned inside _resolve_async, which runs once the task is
actually scheduled. A task that async_executor has created but the loop has not started yet was
therefore invisible to the next _lazy_resolve: it read a _current_task still pointing at an
older, already-finished task, saw .done() was true, and cancelled nothing. Since two
synchronous parameter assignments never yield to the loop, every task queued in that window
escaped cancellation.

The obvious alternative, having _lazy_resolve record the handle it just created, does not work:
async_executor returns None (param/_utils.py:611) and it is a user-replaceable hook that
both Panel and the IPython integration swap out, so changing it to return a task would break
third-party executors.

Changes

Problem 1

The fix separates "what value does this node hold" from "did this node produce a value for the
current inputs". The former stays exactly as it was; only the latter is new, and only propagation
consults it.

rx gains a _skipped flag, set in the except Skip handler and cleared on a successful
resolution. Alongside it, _completed_generation records the generation whose result has actually
landed, and the _awaiting property compares it against _resolve_generation to tell whether an
async resolution is still in flight. _resolve_async stamps _completed_generation at each of the
three sites where it assigns _current_, immediately before triggering, including inside the
async-generator loop, so every yield counts as a resolution.

_resolve() then reports skip state in three places: a node whose _prev is skipped skips too
(rather than computing its operation on a superseded value), a clean node that is awaiting reports
itself skipped while still returning its cached value, and the _shared-reuse branch mirrors the
shared node's skip state.

_rx_transform raises Skip instead of handing over a value when the expression is skipped or
resolves to Skip/Undefined. Skip is already honoured throughout the reference machinery
(_sync_refs, _resolve_ref, _execute_watcher), so watchers, refs and bound functions all
simply do not fire, and fire once with the real value when it lands.

The _shared-reuse branch needed restructuring: it used raise Skip as plain control flow
after assigning self._current_ = self._shared.rx.value, so with _skipped in place the clone
was mislabelled as skipped and a downstream async node then refused to schedule at all. The sync
path now returns its value directly and mirrors self._shared._skipped; only the async path still
raises.

_current (the property Panel's render path reads via rx._callback) is deliberately untouched,
so panes keep displaying the last good value while a recompute is in flight rather than blanking.

Problem 2

A task whose generation has already been superseded now declines to run rather than relying on
being cancelled. The generation counter is incremented synchronously in _lazy_resolve, so by the
time a queued task starts it can always tell whether it is still the current one, and nothing has
awaited obj yet at that point so no body has begun.

The guard runs before self._current_task = task. That ordering matters: a superseded task that
claimed _current_task would clear it again in its finally block, hiding the genuinely live task
from the next _lazy_resolve and reintroducing the same bug by a different route.

Discarding obj is factored into a module-level _close_stale helper, since it has three shapes:
a coroutine closes with obj.close() (which also suppresses the warning Python emits for a
coroutine that is never awaited), an async generator needs await obj.aclose(), and obj is
None on the _shared mirror path, where there is nothing to close. The stale branch of the
async-generator loop, which already did this inline, now calls the same helper.

_lazy_resolve keeps its explicit previous_task.cancel(). The guard only helps tasks that have
not started; cancellation is still what stops a body already suspended at an await. The two
mechanisms are complementary, and there is a test for each.

Behaviour change

A user-raised Skip inside an rx operation no longer notifies watchers with the stale value.

def skip_values(v):
    if v > 2:
        raise Skip
    return v + 1

i = rx(P.param.integer).rx.pipe(skip_values)
items = []
i.rx.watch(items.append)

P.integer = 2   # items == [3]
P.integer = 3   # skipped; on main items becomes [3, 3], now stays [3]

The skip used to re-notify with the previous value. This is the same defect as the async case and what Skip was always meant to do (it's called Skip after all). .rx.value still returns the last value in both cases, so the existing Skip
contract and its tests are unaffected.

Tests

Eleven tests in tests/testreactive.py, all failing before this change.

Problem 1:

• test_reactive_async_watcher_not_notified_while_awaiting
• test_reactive_async_downstream_watcher_not_notified_while_awaiting
• test_reactive_async_downstream_not_computed_while_awaiting, which counts operation invocations
  to show the downstream node is no longer executed on the stale value
• test_reactive_async_rapid_updates_notify_once
• test_reactive_async_ref_not_synced_while_awaiting, covering the allow_refs path
• test_reactive_skip_value_does_not_notify_watcher, pinning the behaviour change above

Problem 2:

• test_reactive_async_superseded_updates_not_computed, asserting which bodies start and which
  finish, rather than timing anything
• test_reactive_async_gen_superseded_updates_not_computed, where a superseded generator is closed
  before its first iteration so its body never runs at all
• test_reactive_async_running_task_is_cancelled, the complementary path: with a yield to the loop
  the body is genuinely suspended and must still be cancelled
• test_async_shared_rx_superseded_updates_computed_once[False, True], a branching pipeline, which
  exercises the obj=None mirror path through the new guard

Full suite: 1532 passed, 4 skipped, 2 xfailed, in both random and fixed test order. No existing
test needed modification. Verified separately that async generators still stream every yield, that
several consumers of one awaiting node are each notified exactly once, and that a chain of async
nodes notifies once with the final value. Panel's test suite is unaffected and mypy is clean on
param/reactive.py.

AI Disclosure

Developed with the assistance of Claude Opus 5

[codecov]

Codecov Report

❌ Patch coverage is 87.80488% with 5 lines in your changes missing coverage. Please review.
✅ Project coverage is 86.90%. Comparing base (51f14ff) to head (e6ccb71).

Files with missing lines	Patch %	Lines
param/reactive.py	87.80%	5 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #1173      +/-   ##
==========================================
+ Coverage   86.73%   86.90%   +0.16%     
==========================================
  Files           9        9              
  Lines        5369     5398      +29     
==========================================
+ Hits         4657     4691      +34     
+ Misses        712      707       -5     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.