fix: Send updates from long running sync callbacks before they return

@@ -13,7 +13,7 @@ import typing as t import weakref -from contextlib import ExitStack, contextmanager+from contextlib import ExitStack, contextmanager, nullcontext from functools import partial, wraps from weakref import WeakKeyDictionary @@ -238,8 +238,10 @@ async def _run_write_futures(doc):     """     Ensure that all write_message calls are awaited and handled.     """+    await _await_writes(_WRITE_FUTURES.pop(doc, []))++async def _await_writes(futures: Iterable[Future]) -> None:     from tornado.websocket import WebSocketClosedError-    futures = _WRITE_FUTURES.pop(doc, [])     for future in futures:         try:             await future@@ -651,6 +653,46 @@ def _suppress_property_callbacks(             event.callback_invoker = None     return events +async def _write_message(+    doc: Document, connections: list[ServerConnection], message: Message+) -> None:+    for conn in connections:+        lock = getattr(conn._socket, 'write_lock', None)+        with (await lock.acquire()) if lock is not None else nullcontext():+            await _await_writes(_socket_dispatcher(conn._socket)(conn, msg=message))++def _write_from_worker_thread(+    doc: Document,+    connections: Iterable[ServerConnection],+    events: list[DocumentChangedEvent]+) -> bool:+    """+    Writes events generated by a locked callback that bokeh runs on a+    worker thread right away, rather than leaving them to bokeh, which+    only writes them once the callback returns, so that a long running+    callback can still report progress.++    Only safe while bokeh has no pending writes of its own: the message+    is serialized here, while the callback holds the Document lock, and+    the write is scheduled on the event loop ahead of bokeh's, so the+    client always receives models before they are referenced.+    """+    patches = [e for e in events if isinstance(e, DocumentPatchedEvent)]+    if state._on_loop_thread or not patches or len(patches) != len(events):+        return False+    loop = state._document_loop(doc)+    asyncio_loop = t.cast(+        'asyncio.AbstractEventLoop | None', getattr(loop, 'asyncio_loop', loop)+    )+    connections = list(connections)+    if asyncio_loop is None or asyncio_loop.is_closed() or not connections:+        return False+    message = connections[0].protocol.create('PATCH-DOC', patches)+    asyncio_loop.call_soon_threadsafe(+        _dispatch_write_task, doc, _write_message, doc, connections, message+    )+    return True+ def _flush_events(curdoc: Document, session: ServerSession) -> None:     """     Dispatches the events held on a Document, either by writing them to@@ -675,6 +717,13 @@ def _flush_events(curdoc: Document, session: ServerSession) -> None:     curdoc.callbacks._held_events = []     monkeypatch_events(events) +    if (+        not queued and session._pending_writes == [] and+        _write_from_worker_thread(curdoc, connections, events)+    ):+        curdoc.unhold()+        return+     # If we cannot write the events ourselves we let bokeh dispatch them,     # as long as it is inside a locked callback and will therefore write     # them before the callback returns. Deferring the write ourselves

@@ -1,5 +1,6 @@ import asyncio import threading+import time  import pytest @@ -17,6 +18,7 @@ from panel.layout import Column from panel.pane import Markdown from panel.tests.util import serve_component, wait_until+from panel.widgets import Button  pytestmark = pytest.mark.ui @@ -144,3 +146,27 @@ async def update():     assert not info['queued']     # The events the hold collected were restored, not dispatched     assert info['held_after'] == info['held_before']+++def test_sync_callback_updates_are_sent_before_it_returns(page):+    """+    Bokeh runs a synchronous event callback on a worker thread and only+    writes its own pending patches once it returns, so updates made+    while it is still running must be written by Panel directly.+    """+    button = Button(label='Click')+    md = Markdown('Idle')++    def cb(event):+        md.object = 'Running'+        time.sleep(2)+        md.object = 'Done'++    button.on_click(cb)++    serve_component(page, Column(button, md))++    page.locator('.bk-btn').click()++    expect(page.locator('.markdown')).to_have_text('Running', timeout=1500)+    expect(page.locator('.markdown')).to_have_text('Done')