fix: publish wallet futures once without blocking HTTP on frames

Author: wock9000

docs/RPC_OPERATIONS.md

@@ -62,6 +62,11 @@ while a newer revision or an expired window refreshes asynchronously; one
 consumer cannot take that snapshot away from another. Blocking wallet reads
 still refresh changed revisions and expired windows. Canonical-branch changes
 invalidate both cached and in-flight publications.
+Cold blocking wallet reads materialize in their existing request thread instead
+of waiting behind unrelated background frames. Concurrent reads of the same
+view still share one future. Its producer assigns and caches the publication
+once, so a waiting request can deliver that exact completed snapshot even if
+a stream consumer has already read it.
 
 The running service commits raw events, cursors, balance jobs, and coalesced
 accounting jobs atomically. A separate accounting worker replays each affected

src/rhpools/lp_market_service.py

@@ -2742,6 +2742,24 @@ def _fresh_owner_envelope(
         }
 
 
+    def _materialize_owner_projection(
+            self, params: Mapping[str, Any],
+    ) -> tuple[tuple[int, int, int, int], float | None, dict[str, Any]]:
+        version, valid_until, envelope = self._owners_result(params)
+        view_key = self._owner_view_key(params)
+        with self._frame_lock:
+            self._owner_result_revision += 1
+            envelope = {**envelope, "revision": self._owner_result_revision}
+            ready = (version, valid_until, envelope)
+            self._owner_results[view_key] = ready
+            self._owner_results.move_to_end(view_key)
+            self._owner_last_started[view_key] = time.monotonic()
+            while len(self._owner_results) > 64:
+                stale_view, _ = self._owner_results.popitem(last=False)
+                if ("owners", stale_view) not in self._frame_futures:
+                    self._owner_last_started.pop(stale_view, None)
+        return ready
+
     def _owner_projection(
             self, raw_params: Mapping[str, Any], *, wait: bool,
     ) -> dict[str, Any] | None:
@@ -2757,29 +2775,17 @@ def _owner_projection(
             ] | None = None
             pending: Future | None = None
             delay = 0.0
+            leader = False
             failure: BaseException | None = None
             with self._frame_lock:
                 pending = self._frame_futures.get(future_key)
                 if pending is not None and pending.done():
                     self._frame_futures.pop(future_key, None)
                     try:
-                        version, valid_until, envelope = pending.result()
+                        ready = pending.result()
                     except BaseException as exc:
                         failure = exc
                     else:
-                        self._owner_result_revision += 1
-                        envelope = {
-                            **envelope,
-                            "revision": self._owner_result_revision,
-                        }
-                        ready = (version, valid_until, envelope)
-                        self._owner_results[view_key] = ready
-                        self._owner_results.move_to_end(view_key)
-                        self._owner_last_started[view_key] = time.monotonic()
-                        while len(self._owner_results) > 64:
-                            stale_view, _ = self._owner_results.popitem(last=False)
-                            if ("owners", stale_view) not in self._frame_futures:
-                                self._owner_last_started.pop(stale_view, None)
                         pending = None
                 if failure is None:
                     cached = self._owner_results.get(view_key)
@@ -2810,13 +2816,30 @@ def _owner_projection(
                         earliest = self._owner_last_started.get(view_key, 0.0) + 1.0
                         delay = max(0.0, earliest - now)
                         if delay == 0.0:
-                            pending = self._frame_executor.submit(
-                                self._owners_result, dict(params),
-                            )
+                            if wait and ready is None:
+                                pending = Future()
+                                leader = True
+                            else:
+                                pending = self._frame_executor.submit(
+                                    self._materialize_owner_projection, dict(params),
+                                )
                             self._frame_futures[future_key] = pending
                             self._owner_last_started[view_key] = now
             if failure is not None:
                 raise failure
+            if leader:
+                assert pending is not None
+                try:
+                    pending.set_result(self._materialize_owner_projection(params))
+                except BaseException as exc:
+                    pending.set_exception(exc)
+            if ready is None and wait and pending is not None:
+                try:
+                    ready = pending.result()
+                finally:
+                    with self._frame_lock:
+                        if self._frame_futures.get(future_key) is pending:
+                            self._frame_futures.pop(future_key, None)
             if ready is not None:
                 version, _valid_until, envelope = ready
                 # Continuous appends cannot starve completed, qualified snapshots.
@@ -2832,15 +2855,6 @@ def _owner_projection(
                 return fresh
             if not wait:
                 return None
-            if pending is not None:
-                try:
-                    pending.result()
-                except BaseException:
-                    with self._frame_lock:
-                        if self._frame_futures.get(future_key) is pending:
-                            self._frame_futures.pop(future_key, None)
-                    raise
-                continue
             if delay:
                 time.sleep(delay)
 

tests/test_lp_market_service.py

@@ -1905,6 +1905,48 @@ def read_feed():
         app.close()
 
 
+def test_wallet_read_does_not_queue_behind_unrelated_frames(tmp_path, monkeypatch):
+    app = service(tmp_path / "market.sqlite")
+    release = threading.Event()
+    entered = [threading.Event(), threading.Event()]
+    returned = threading.Event()
+    result = {}
+    reader = None
+    original = app.frame
+
+    def slow_frame(params, *args, **kwargs):
+        entered[int(params["q"])].set()
+        assert release.wait(3)
+        return original(params, *args, **kwargs)
+
+    monkeypatch.setattr(app, "frame", slow_frame)
+    try:
+        block = header(100, int(time.time()))
+        event = lp_effect(
+            block, "v4", "add", 1_000, (1_000_000, 1_000_000),
+            position_state(0), position_state(1_000),
+        )
+        app.observe_current_block(block)
+        app.observe_current_events(block, (event,))
+        for index in range(2):
+            assert app.poll_frame({"window": "all", "q": str(index)}, -1, 0) is None
+            assert entered[index].wait(1)
+
+        def read_wallet():
+            result.update(app.owners({"window": "all", "q": TOKEN}))
+            returned.set()
+
+        reader = threading.Thread(target=read_wallet)
+        reader.start()
+        assert returned.wait(1), "wallet read queued behind unrelated frame SQL"
+        assert TOKEN in {row["owner"] for row in result["rows"]}
+    finally:
+        release.set()
+        if reader is not None:
+            reader.join(3)
+        app.close()
+
+
 @pytest.mark.parametrize("reorg", [False, True])
 def test_completed_wallet_projection_is_deliverable_during_live_changes(
         tmp_path, monkeypatch, reorg):