From 76735189de38b319b2deb2d3209e54bf38759746 Mon Sep 17 00:00:00 2001 From: Tyler Goodlet Date: Sat, 9 Nov 2024 15:14:34 -0500 Subject: [PATCH 1/3] data._web_bs: try to raise jsonrpc errors in parent task --- piker/data/_web_bs.py | 67 ++++++++++++++++++++++++++++++++----------- 1 file changed, 50 insertions(+), 17 deletions(-) diff --git a/piker/data/_web_bs.py b/piker/data/_web_bs.py index 256b35af..a401588d 100644 --- a/piker/data/_web_bs.py +++ b/piker/data/_web_bs.py @@ -273,7 +273,7 @@ async def _reconnect_forever( nobsws._connected.set() await trio.sleep_forever() except HandshakeError: - log.exception(f'Retrying connection') + log.exception('Retrying connection') # ws & nursery block ends @@ -359,8 +359,8 @@ async def open_autorecon_ws( ''' -JSONRPC response-request style machinery for transparent multiplexing of msgs -over a NoBsWs. +JSONRPC response-request style machinery for transparent multiplexing +of msgs over a NoBsWs. ''' @@ -377,16 +377,20 @@ async def open_jsonrpc_session( url: str, start_id: int = 0, response_type: type = JSONRPCResult, - request_type: Optional[type] = None, - request_hook: Optional[Callable] = None, - error_hook: Optional[Callable] = None, + # request_type: Optional[type] = None, + # request_hook: Optional[Callable] = None, + # error_hook: Optional[Callable] = None, ) -> Callable[[str, dict], dict]: + # NOTE, store all request msgs so we can raise errors on the + # caller side! + req_msgs: dict[int, dict] = {} + async with ( trio.open_nursery() as n, open_autorecon_ws(url) as ws ): - rpc_id: Iterable = count(start_id) + rpc_id: Iterable[int] = count(start_id) rpc_results: dict[int, dict] = {} async def json_rpc(method: str, params: dict) -> dict: @@ -394,26 +398,40 @@ async def open_jsonrpc_session( perform a json rpc call and wait for the result, raise exception in case of error field present on response ''' + nonlocal req_msgs + + req_id: int = next(rpc_id) msg = { 'jsonrpc': '2.0', - 'id': next(rpc_id), + 'id': req_id, 'method': method, 'params': params } _id = msg['id'] - rpc_results[_id] = { + result = rpc_results[_id] = { 'result': None, - 'event': trio.Event() + 'error': None, + 'event': trio.Event(), # signal caller resp arrived } + req_msgs[_id] = msg await ws.send_msg(msg) + # wait for reponse before unblocking requester code await rpc_results[_id]['event'].wait() - ret = rpc_results[_id]['result'] + if (maybe_result := result['result']): + ret = maybe_result + del rpc_results[_id] - del rpc_results[_id] + else: + err = result['error'] + raise Exception( + f'JSONRPC request failed\n' + f'req: {msg}\n' + f'resp: {err}\n' + ) if ret.error is not None: raise Exception(json.dumps(ret.error, indent=4)) @@ -428,6 +446,7 @@ async def open_jsonrpc_session( the server side. ''' + nonlocal req_msgs async for msg in ws: match msg: case { @@ -451,15 +470,29 @@ async def open_jsonrpc_session( 'params': _, }: log.debug(f'Recieved\n{msg}') - if request_hook: - await request_hook(request_type(**msg)) + # if request_hook: + # await request_hook(request_type(**msg)) case { 'error': error }: - log.warning(f'Recieved\n{error}') - if error_hook: - await error_hook(response_type(**msg)) + # if error_hook: + # await error_hook(response_type(**msg)) + + # retreive orig request msg, set error + # response in original "result" msg, + # THEN FINALLY set the event to signal caller + # to raise the error in the parent task. + req_id: int = error['id'] + req_msg: dict = req_msgs[req_id] + result: dict = rpc_results[req_id] + result['error'] = error + result['event'].set() + log.error( + f'JSONRPC request failed\n' + f'req: {req_msg}\n' + f'resp: {error}\n' + ) case _: log.warning(f'Unhandled JSON-RPC msg!?\n{msg}') -- 2.34.1 From 5633f5614d8c292c034c20545e20f5b488d65c7e Mon Sep 17 00:00:00 2001 From: Tyler Goodlet Date: Wed, 12 Feb 2025 11:07:27 -0500 Subject: [PATCH 2/3] Doc-n-clean `.data._web_bs.open_jsonrpc_session()` Add a doc-string reflecting recent refinements, drop all the old hook params, rename `n: trio.Nursery` -> `tn` for "task nursery" fitting with code base's naming style. --- piker/data/_web_bs.py | 22 +++++++++++----------- 1 file changed, 11 insertions(+), 11 deletions(-) diff --git a/piker/data/_web_bs.py b/piker/data/_web_bs.py index a401588d..e9d2deeb 100644 --- a/piker/data/_web_bs.py +++ b/piker/data/_web_bs.py @@ -377,17 +377,22 @@ async def open_jsonrpc_session( url: str, start_id: int = 0, response_type: type = JSONRPCResult, - # request_type: Optional[type] = None, - # request_hook: Optional[Callable] = None, - # error_hook: Optional[Callable] = None, ) -> Callable[[str, dict], dict]: + ''' + Init a json-RPC-over-websocket connection to the provided `url`. + A `json_rpc: Callable[[str, dict], dict` is delivered to the + caller for sending requests and a bg-`trio.Task` handles + processing of response msgs including error reporting/raising in + the parent/caller task. + + ''' # NOTE, store all request msgs so we can raise errors on the # caller side! req_msgs: dict[int, dict] = {} async with ( - trio.open_nursery() as n, + trio.open_nursery() as tn, open_autorecon_ws(url) as ws ): rpc_id: Iterable[int] = count(start_id) @@ -470,15 +475,10 @@ async def open_jsonrpc_session( 'params': _, }: log.debug(f'Recieved\n{msg}') - # if request_hook: - # await request_hook(request_type(**msg)) case { 'error': error }: - # if error_hook: - # await error_hook(response_type(**msg)) - # retreive orig request msg, set error # response in original "result" msg, # THEN FINALLY set the event to signal caller @@ -497,6 +497,6 @@ async def open_jsonrpc_session( case _: log.warning(f'Unhandled JSON-RPC msg!?\n{msg}') - n.start_soon(recv_task) + tn.start_soon(recv_task) yield json_rpc - n.cancel_scope.cancel() + tn.cancel_scope.cancel() -- 2.34.1 From e391c896f847dda05f45f95dfdf981273a7ebacf Mon Sep 17 00:00:00 2001 From: Tyler Goodlet Date: Tue, 19 Nov 2024 16:58:40 -0500 Subject: [PATCH 3/3] Mk jsronrpc's underlying ws timeout `float('inf')` Since currently we're only using this IPC subsys for `deribit`, and generally speaking we're primarly supporting options markets (which are fairly "slow moving"), flip to a default of NOT resetting the `NoBsWs` on timeout since doing so normally breaks the jsron-rpc IPC session. Without a proper `fixture` passed to `open_autorecon_ws()` (which we should eventually implement!!) relying on a timeout-to-reset more or less will just cause breakage issues - a proper reconnect sequence must be implemented before using that feature. Deats, - expose and proxy through the `msg_recv_timeout` from `open_jsonrpc_session()` into the underlying `open_autorecon_ws()` call. --- piker/data/_web_bs.py | 22 +++++++++++++++++++--- 1 file changed, 19 insertions(+), 3 deletions(-) diff --git a/piker/data/_web_bs.py b/piker/data/_web_bs.py index e9d2deeb..4d886fbc 100644 --- a/piker/data/_web_bs.py +++ b/piker/data/_web_bs.py @@ -360,7 +360,7 @@ async def open_autorecon_ws( ''' JSONRPC response-request style machinery for transparent multiplexing -of msgs over a NoBsWs. +of msgs over a `NoBsWs`. ''' @@ -377,6 +377,16 @@ async def open_jsonrpc_session( url: str, start_id: int = 0, response_type: type = JSONRPCResult, + msg_recv_timeout: float = float('inf'), + # ^NOTE, since only `deribit` is using this jsonrpc stuff atm + # and options mkts are generally "slow moving".. + # + # FURTHER if we break the underlying ws connection then since we + # don't pass a `fixture` to the task that manages `NoBsWs`, i.e. + # `_reconnect_forever()`, the jsonrpc "transport pipe" get's + # broken and never restored with wtv init sequence is required to + # re-establish a working req-resp session. + ) -> Callable[[str, dict], dict]: ''' Init a json-RPC-over-websocket connection to the provided `url`. @@ -393,12 +403,18 @@ async def open_jsonrpc_session( async with ( trio.open_nursery() as tn, - open_autorecon_ws(url) as ws + open_autorecon_ws( + url=url, + msg_recv_timeout=msg_recv_timeout, + ) as ws ): rpc_id: Iterable[int] = count(start_id) rpc_results: dict[int, dict] = {} - async def json_rpc(method: str, params: dict) -> dict: + async def json_rpc( + method: str, + params: dict, + ) -> dict: ''' perform a json rpc call and wait for the result, raise exception in case of error field present on response -- 2.34.1