From 2aa66c2cf6c76b8e270af9367d27ca16f63e7024 Mon Sep 17 00:00:00 2001 From: goodboy Date: Tue, 11 Aug 2026 18:46:23 -0400 Subject: [PATCH] Fix IB `MethodProxy` response routing `open_client_proxy()` broadcast its one-slot linked channel to an always-running status relay and an idle method receiver. Status traffic advanced only the relay, so the first lazy contract lookup for an uncached order raised `Lagged` before receiving its response. Give `relay_client_proxy_messages()` sole receive ownership. Route tagged responses from concurrent caller tasks into their per-`mid` slots while each proxy context retains its own channel and relay. Deats, - log stale responses with their original method name; - warn when unresolved-request metadata exceeds its bounded limit; - drop late responses after caller cancellation; - preserve the reconnect design note and message-case rationale; - accept clean channel EOF during proxy teardown; - cover idle status, concurrency, cancellation and EOF paths. Prompt-IO: ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.md (this patch was generated in some part by `opencode` using `gpt-5.6-sol` (`openai`)) --- .../run-tests/test-harness-reference.md | 3 + .../20260811T205826Z_06d5ea5d_prompt_io.md | 36 +++ ...20260811T205826Z_06d5ea5d_prompt_io.raw.md | 66 +++++ piker/brokers/ib/api.py | 224 ++++++++++----- tests/test_ib_method_proxy.py | 271 ++++++++++++++++++ 5 files changed, 523 insertions(+), 77 deletions(-) create mode 100644 ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.md create mode 100644 ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.raw.md create mode 100644 tests/test_ib_method_proxy.py diff --git a/.claude/skills/run-tests/test-harness-reference.md b/.claude/skills/run-tests/test-harness-reference.md index 9d2d0c3b..dc419d64 100644 --- a/.claude/skills/run-tests/test-harness-reference.md +++ b/.claude/skills/run-tests/test-harness-reference.md @@ -108,6 +108,7 @@ Deterministic or local first-pass targets: - `tests/test_store_cli.py` - `tests/test_backfill_audit_snippet.py` - `tests/test_ib_history.py` +- `tests/test_ib_method_proxy.py` - `tests/test_history_backfill.py` - `tests/test_ldshm.py` - `tests/test_accounting.py::test_account_file_default_empty` @@ -193,6 +194,7 @@ tests/ test_ems.py actor, EMS, and paper-position behavior test_feeds.py live Binance/Kraken feeds and shared memory test_ib_history.py deterministic IB history request formatting + test_ib_method_proxy.py deterministic IB asyncio proxy routing test_history_backfill.py deterministic history/SHM orchestration test_ldshm.py SHM unpublished-slot guard test_questrade.py obsolete credentialed tests; skipped @@ -213,6 +215,7 @@ tests/ | `piker/storage/cli.py` command UX | `tests/test_store_cli.py` | fake SHM/runtime, no mutation | | `snippets/nativedb_backfill_audit.xsh` | `tests/test_backfill_audit_snippet.py` | disposable paths only | | `piker/brokers/ib/api.py`, `feed.py` history | `tests/test_ib_history.py` | fake client, no network | +| `piker/brokers/ib/api.py` method proxy | `tests/test_ib_method_proxy.py` | fake channel, no network | | `piker/tsp/_history.py` | `tests/test_history_backfill.py` | fake provider/storage/SHM | | `piker/storage/cli.py` SHM null-slot guard | `tests/test_ldshm.py` | synthetic timestamps, no SHM mutation | | `piker/config.py` | `test_account_file_default_empty` | root-network test has a known mismatch | diff --git a/ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.md b/ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.md new file mode 100644 index 00000000..9af9b3ea --- /dev/null +++ b/ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.md @@ -0,0 +1,36 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +session: ses_0799212ebffe42arY96czXn89F +timestamp: 2026-08-11T20:58:26Z +git_ref: 06d5ea5d +scope: code +substantive: true +raw_file: 20260811T205826Z_06d5ea5d_prompt_io.raw.md +--- + +## Prompt + +Diagnose and fix an IB live-order failure where lazy NVDA contract +qualification raised `tractor.trionics.Lagged` after the order reached +`brokerd.ib`. Determine whether recent history changes or cancelled symbol +searches caused the stale state. + +## Response summary + +Traced the failure to competing consumers of a one-slot Trio broadcast, not +the recent SHM change or stale asyncio client state. Replaced broadcast +fan-out with one response/status demultiplexer, retained request-ID +correlation, handled cancellation and teardown, documented the task/channel +ownership model, warned on diagnostic metadata eviction, and added +deterministic proxy regressions. + +## Files changed + +- `piker/brokers/ib/api.py` - sole-reader IB proxy response routing. +- `tests/test_ib_method_proxy.py` - status, concurrency, cancellation and EOF regressions. +- `.claude/skills/run-tests/test-harness-reference.md` - proxy test mapping. + +## Human edits + +None - generated output follows the user's live failure diagnosis. diff --git a/ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.raw.md b/ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.raw.md new file mode 100644 index 00000000..ba7c41b0 --- /dev/null +++ b/ai/prompt-io/opencode/20260811T205826Z_06d5ea5d_prompt_io.raw.md @@ -0,0 +1,66 @@ +--- +model: openai/gpt-5.6-sol +service: opencode +timestamp: 2026-08-11T20:58:26Z +git_ref: 06d5ea5d +diff_cmd: git diff HEAD~1..HEAD +--- + +The user reported that certain IB order entries might have broken and +provided a live failure for a first NVDA sell order. EMS delivered the +request to `brokerd.ib`, but lazy `get_mkt_info('nvda.nasdaq')` contract +qualification failed in `MethodProxy._run_method()` with: + +```text +tractor.trionics._broadcast.Lagged: +Task `piker.brokers.ib.broker.handle_order_requests` overrun and +dropped `0` values +``` + +The user suspected a cancelled symbol-search task might have left stale +asyncio state. Cancellation can leave an orphaned method response, but the +observed failure was stale Trio broadcast state. `open_client_proxy()` +broadcast a one-slot `LinkedTaskChannel` to an always-running IB status relay +and an idle method receiver. Status traffic advanced only the relay, so the +first method call for an uncached symbol entered an already-lagged receiver. +Orders for eagerly cached contracts bypassed this path. + +The immediately preceding sparse-history commit did not modify IB order, +symbol, or proxy code; it exposed no direct dependency on this failure. + +> `git diff HEAD~1..HEAD -- piker/brokers/ib/api.py` + +Generated a sole-reader message demultiplexer for each IB client proxy. +`relay_client_proxy_messages()` owns every asyncio-to-Trio receive, handles +status/error traffic, and dispatches `mid`-tagged responses into per-call +`MethodProxy` wait slots. Cancelled calls remove their pending slot, causing +late responses to be logged and dropped without skewing later calls. Clean +channel EOF is accepted during normal proxy teardown. + +Follow-up review restored the masked reconnect TODO and per-message case +rationale, documented the one-reader/many-writer task hierarchy and separate +per-proxy channels, and added a warning with request method metadata when the +bounded unresolved-request diagnostic table evicts its oldest entry. + +> `git diff HEAD~1..HEAD -- tests/test_ib_method_proxy.py` + +Generated deterministic regressions for idle status traffic before first +qualification, two concurrent calls with reverse-order responses, cancelled +calls with late responses followed by a successful call, and graceful +channel EOF. Every synchronization wait is timeout-bounded. + +> `git diff HEAD~1..HEAD -- .claude/skills/run-tests/test-harness-reference.md` + +Registered the deterministic IB method-proxy target and source mapping. + +Verification output: + +```text +........................ [100%] +25 passed, 4 warnings in 3.44s +``` + +The warnings are existing Tractor/Trio deprecations from the EMS test. +Adversarial review drove a clean-EOF guard and bounded polling. Final review +found no actionable issues. Live IB order validation remains pending and +requires restarting `brokerd.ib`. diff --git a/piker/brokers/ib/api.py b/piker/brokers/ib/api.py index 496c9d18..6e0bd6ed 100644 --- a/piker/brokers/ib/api.py +++ b/piker/brokers/ib/api.py @@ -1643,6 +1643,61 @@ class MethodProxy: # to the next caller causing an off-by-one skew of # every result thereafter! self._mids = itertools.count() + self._pending: dict[ + int, + tuple[trio.Event, list[dict]], + ] = {} + self._request_methods: dict[int, str|None] = {} + + def _deliver_method_response( + self, + msg: dict, + + ) -> bool: + ''' + Deliver one correlated response to its waiting caller. + + ''' + mid: int = msg['mid'] + meth: str|None = self._request_methods.pop(mid, None) + pending = self._pending.get(mid) + if pending is None: + log.warning( + f'Dropping stale method-resp,\n' + f'mid: {mid}\n' + f'meth: {meth!r}\n' + f'(caller prolly got cancelled ' + f'before its resp?)\n' + ) + return False + + event, slot = pending + slot.append(msg) + event.set() + return True + + def _track_request_method( + self, + mid: int, + meth: str|None, + + ) -> None: + ''' + Retain bounded request metadata for stale-response logs. + + ''' + self._request_methods[mid] = meth + if len(self._request_methods) > 256: + oldest_mid: int = next(iter(self._request_methods)) + oldest_meth: str|None = self._request_methods.pop( + oldest_mid + ) + log.warning( + f'Evicting unresolved method-request metadata,\n' + f'mid: {oldest_mid}\n' + f'meth: {oldest_meth!r}\n' + f'pending metadata limit: 256\n' + ) async def _run_method( self, @@ -1656,65 +1711,30 @@ class MethodProxy: ``tractor.to_asyncio`` layer. ''' - chan = self.chan mid: int = next(self._mids) - await chan.send((meth, kwargs, mid)) + event = trio.Event() + slot: list[dict] = [] + self._pending[mid] = event, slot + self._track_request_method(mid, meth) + try: + await self.chan.send((meth, kwargs, mid)) + await event.wait() + finally: + self._pending.pop(mid, None) - while not chan.closed(): - msg = await chan.receive() + msg: dict = slot[0] + match msg: + # OUR method-call response B) + case {'result': res}: + return res - # TODO: implement reconnect functionality like - # in our `.data._web_bs.NoBsWs` - # try: - # msg = await chan.receive() - # except ConnectionError: - # self.reset() + case {'exception': err}: + raise err - match msg: - # OUR method-call response B) - case {'mid': resp_mid, 'result': res} if ( - resp_mid == mid - ): - return res - - case {'mid': resp_mid, 'exception': err} if ( - resp_mid == mid - ): - raise err - - # an "orphaned" response to some prior - # (cancelled) caller's request; drop it and - # keep waiting for ours. - case {'mid': resp_mid}: - log.warning( - f'Dropping stale method-resp,\n' - f'mid: {resp_mid} (ours: {mid})\n' - f'(a prior caller prolly got ' - f'cancelled before its resp?)\n' - ) - continue - - # out-of-band (inline) client error: raise - # to the current caller as before. - case {'exception': err}: - raise err - - case ('error', emsg): - log.warning(f'IB error relay: {emsg}') - continue - - # routine (api-farm conn) status events - # relayed inline by `Client.inline_errors()`; - # not a response to our method call so just - # log at info and keep waiting. - case ('event', emsg): - log.info( - f'IB status event relay: {emsg}' - ) - continue - - case _: - log.warning(f'UNKNOWN IB MSG: {msg}') + case _: + raise RuntimeError( + f'Invalid method response for mid={mid}: {msg!r}' + ) def status_event( self, @@ -1800,6 +1820,73 @@ async def open_aio_client_method_relay( raise ValueError(f'Unhandled msg {msg}') +async def relay_client_proxy_messages( + chan: tractor.to_asyncio.LinkedTaskChannel, + proxy: MethodProxy, + +) -> None: + ''' + Route one proxy's asyncio-client msgs from a single reader. + + Each `open_client_proxy()` allocates its own linked channel and + starts this task in its relay nursery. Any number of Trio caller + tasks may concurrently write tagged requests through + `MethodProxy._run_method()` and wait on their per-`mid` slots. + This task alone reads the channel and dispatches each response to + its waiting caller. + + Separate proxy contexts own separate channels and reader tasks; + no cross-proxy response broadcast is intended. IB status events + still reach each proxy through its own `Client.inline_errors()` + handler. + + ''' + try: + while not chan.closed(): + # TODO: implement reconnect functionality like + # in our `.data._web_bs.NoBsWs` + # try: + msg = await chan.receive() + # except ConnectionError: + # proxy.reset() + + match msg: + # Correlated response from a method call. + case {'mid': _}: + proxy._deliver_method_response(msg) + + # routine (api-farm conn) status events + # relayed inline by `Client.inline_errors()`; + # not a response to a method call so just + # log at info and keep waiting. + case ('event', status_msg): + reason = status_msg['reason'] + event = proxy.event_table.pop(reason, None) + if ( + event + and + event.statistics().tasks_waiting + ): + log.info(f'Relaying ib status message: {msg}') + event.set() + + # Inline API error with no request correlation. + case ('error', emsg): + log.warning(f'IB error relay: {emsg}') + + # Out-of-band client error with no method `mid`. + case {'exception': err}: + log.error( + f'Uncorrelated IB client error: {err!r}' + ) + + # Preserve visibility into unexpected relay traffic. + case _: + log.warning(f'UNKNOWN IB MSG: {msg}') + except trio.EndOfChannel: + log.info('IB client proxy relay closed') + + @acm async def open_client_proxy( client: Client, @@ -1836,28 +1923,11 @@ async def open_client_proxy( continue setattr(proxy, name, partial(proxy._run_method, meth=name)) - async def relay_events(): - - async with chan.subscribe() as msg_stream: - - async for msg in msg_stream: - if 'event' not in msg: - continue - - # if 'event' in msg: - # wake up any system event waiters. - etype, status_msg = msg - reason = status_msg['reason'] - - ev = proxy.event_table.pop(reason, None) - - if ev and ev.statistics().tasks_waiting: - log.info(f'Relaying ib status message: {msg}') - ev.set() - - continue - - relay_tn.start_soon(relay_events) + relay_tn.start_soon( + relay_client_proxy_messages, + chan, + proxy, + ) yield proxy diff --git a/tests/test_ib_method_proxy.py b/tests/test_ib_method_proxy.py new file mode 100644 index 00000000..051a7e59 --- /dev/null +++ b/tests/test_ib_method_proxy.py @@ -0,0 +1,271 @@ +''' +IB asyncio method-proxy regressions. + +''' +from types import SimpleNamespace +from typing import Any + +import pytest +import trio + +from piker.brokers.ib import api as ib_api +from piker.brokers.ib.api import ( + MethodProxy, + relay_client_proxy_messages, +) + + +class FakeChannel: + ''' + Model the Trio side of an asyncio linked-task channel. + + ''' + def __init__(self) -> None: + self.requests: list[tuple] = [] + ( + self._tx, + self._rx, + ) = trio.open_memory_channel[Any](10) + + async def send(self, msg: tuple) -> None: + ''' + Capture one Trio-to-asyncio method request. + + ''' + self.requests.append(msg) + + async def receive(self) -> Any: + ''' + Receive one simulated asyncio-to-Trio response. + + ''' + return await self._rx.receive() + + def closed(self) -> bool: + ''' + Report an open channel for the lifetime of each test. + + ''' + return False + + async def respond(self, msg: Any) -> None: + ''' + Send one response or status event to the proxy relay. + + ''' + await self._tx.send(msg) + + async def close_responses(self) -> None: + ''' + Close the simulated asyncio-to-Trio response channel. + + ''' + await self._tx.aclose() + + +def test_method_proxy_routes_after_idle_status_events() -> None: + ''' + Idle IB status traffic must not break first-symbol qualification. + + ``open_client_proxy()`` previously broadcast a one-slot channel + to an always-running status relay and an idle method receiver. + Status events advanced only the relay, so a first order for an + uncached symbol raised ``Lagged`` before receiving + ``get_sym_details()``. Feed several idle events through the + sole-reader relay, then start two method calls and return their + responses in reverse order. Both exact results prove status + traffic cannot overrun a dormant caller and concurrent responses + remain correlated by request ID. + + ''' + async def main() -> None: + chan = FakeChannel() + proxy = MethodProxy( + chan, + event_table={}, + asyncio_ns=SimpleNamespace(), + ) + results: dict[str, str] = {} + + async def call(name: str) -> None: + results[name] = await proxy._run_method(meth=name) + + async with trio.open_nursery() as nursery: + nursery.start_soon( + relay_client_proxy_messages, + chan, + proxy, + ) + for i in range(3): + await chan.respond(( + 'event', + {'reason': f'farm-status-{i}'}, + )) + + nursery.start_soon(call, 'first') + nursery.start_soon(call, 'second') + with trio.fail_after(0.1): + while len(chan.requests) < 2: + await trio.lowlevel.checkpoint() + + mids: dict[str, int] = { + meth: mid + for meth, _kwargs, mid in chan.requests + } + await chan.respond({ + 'mid': mids['second'], + 'result': 'second-result', + }) + await chan.respond({ + 'mid': mids['first'], + 'result': 'first-result', + }) + with trio.fail_after(0.1): + while len(results) < 2: + await trio.lowlevel.checkpoint() + nursery.cancel_scope.cancel() + + assert results == { + 'first': 'first-result', + 'second': 'second-result', + } + + trio.run(main) + + +def test_method_proxy_relay_accepts_clean_eof() -> None: + ''' + Normal asyncio relay completion must not fail proxy teardown. + + The sole-reader relay blocks in ``LinkedTaskChannel.receive()`` + until the asyncio task exits and closes its Trio send channel. + Close the fake response sender before entering the relay and + prove the resulting ``EndOfChannel`` is graceful completion. + + ''' + async def main() -> None: + chan = FakeChannel() + proxy = MethodProxy( + chan, + event_table={}, + asyncio_ns=SimpleNamespace(), + ) + await chan.close_responses() + with trio.fail_after(0.1): + await relay_client_proxy_messages(chan, proxy) + + trio.run(main) + + +def test_method_proxy_drops_cancelled_call_response() -> None: + ''' + A cancelled symbol search must not skew the next method response. + + Search timeouts can cancel a Trio caller after its request + reaches the asyncio client. Wait until one request is sent, + cancel its caller, and then deliver the orphaned response. The + sole-reader relay must drop that stale ``mid`` without failing; a + subsequent call must still receive its own exact response. + + ''' + async def main() -> None: + chan = FakeChannel() + proxy = MethodProxy( + chan, + event_table={}, + asyncio_ns=SimpleNamespace(), + ) + call_scope = trio.CancelScope() + cancelled = trio.Event() + result: list[str] = [] + + async def cancelled_call() -> None: + with call_scope: + try: + await proxy._run_method(meth='cancelled') + finally: + cancelled.set() + + async def next_call() -> None: + value = await proxy._run_method(meth='next') + result.append(value) + + async with trio.open_nursery() as nursery: + nursery.start_soon( + relay_client_proxy_messages, + chan, + proxy, + ) + nursery.start_soon(cancelled_call) + with trio.fail_after(0.1): + while len(chan.requests) < 1: + await trio.lowlevel.checkpoint() + cancelled_mid: int = chan.requests[0][2] + call_scope.cancel() + with trio.fail_after(0.1): + await cancelled.wait() + assert proxy._pending == {} + assert ( + proxy._request_methods[cancelled_mid] + == + 'cancelled' + ) + + await chan.respond({ + 'mid': cancelled_mid, + 'result': 'stale', + }) + with trio.fail_after(0.1): + while cancelled_mid in proxy._request_methods: + await trio.lowlevel.checkpoint() + nursery.start_soon(next_call) + with trio.fail_after(0.1): + while len(chan.requests) < 2: + await trio.lowlevel.checkpoint() + next_mid: int = chan.requests[1][2] + await chan.respond({ + 'mid': next_mid, + 'result': 'fresh', + }) + with trio.fail_after(0.1): + while not result: + await trio.lowlevel.checkpoint() + nursery.cancel_scope.cancel() + + assert result == ['fresh'] + + trio.run(main) + + +def test_method_proxy_warns_when_metadata_is_evicted( + monkeypatch: pytest.MonkeyPatch, +) -> None: + ''' + Unresolved metadata eviction must remain operator-visible. + + Cancelled calls whose asyncio responses never arrive retain a + bounded method name solely for later stale-response diagnostics. + Fill that table beyond its limit and prove the oldest entry is + evicted with its request ID, method name, and configured bound in + the warning. + + ''' + proxy = MethodProxy( + FakeChannel(), + event_table={}, + asyncio_ns=SimpleNamespace(), + ) + warnings: list[str] = [] + monkeypatch.setattr(ib_api.log, 'warning', warnings.append) + + for mid in range(257): + proxy._track_request_method(mid, f'method-{mid}') + + assert len(proxy._request_methods) == 256 + assert 0 not in proxy._request_methods + assert warnings == [ + 'Evicting unresolved method-request metadata,\n' + 'mid: 0\n' + "meth: 'method-0'\n" + 'pending metadata limit: 256\n' + ]