pikers
/
piker
4
1
Fork 0
Code Issues 20 Pull Requests 11 Packages Projects Releases Wiki Activity

Replace `TooFastEdit` sentinel with `set` tracker 

Drop the pattern of storing a `TooFastEdit` exc
instance in `reqids2txids` as a sentinel value;
instead track affected reqids in a dedicated
`toofastedit: set[int]` and check membership
via `reqid in toofastedit`.

Deats,
- Comment out `TooFastEdit` class and its `reg_err_types()` call.
- Add `toofastedit` param to both `handle_order_requests()` and
  `handle_order_updates()`, threaded from `open_trade_dialog()`.

Also,
- Use `partial()` with kwargs for the `tn.start_soon()` call to the
  order handler.
- Add `await tractor.pause()` on the too-fast edit path for runtime
  debugging; will remove once confident this all works.
- Expand comments explaining the cancel/edit race condition.

(this commit msg was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code
Gud Boi 2026-04-15 19:04:17 -04:00
parent 8cefc1bdf8
commit d06bbec24f
1 changed files with 43 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

61
piker/brokers/kraken/broker.py
View File

@ -97,11 +97,11 @@ MsgUnion = Union[
] ]
class TooFastEdit(Exception): # class TooFastEdit(Exception):
'Edit requests faster then api submissions' # 'Edit requests faster then api submissions'
reg_err_types([TooFastEdit]) # reg_err_types([TooFastEdit])
# TODO: make this wrap the `api.Client` and `ws` instances # TODO: make this wrap the `api.Client` and `ws` instances
@ -138,11 +138,12 @@ async def handle_order_requests(
apiflows: OrderDialogs, apiflows: OrderDialogs,
ids: bidict[str, int], ids: bidict[str, int],
reqids2txids: dict[int, str], reqids2txids: dict[int, str],
toofastedit: set[int],
) -> None: ) -> None:
''' '''
Process new order submission requests from the EMS `trio.Task` which handles order ctl requests from the EMS and
and deliver acks or errors. deliver acks or errors back on that IPC dialog.
''' '''
# XXX: UGH, let's unify this.. with ``msgspec``!!! # XXX: UGH, let's unify this.. with ``msgspec``!!!
@ -160,8 +161,14 @@ async def handle_order_requests(
txid = reqids2txids[reqid] txid = reqids2txids[reqid]
except KeyError: except KeyError:
# XXX: not sure if this block ever gets hit now? # XXX: not sure if this block ever gets hit now?
# SEEMS TO on the race case with the update task?
# - update dark order quickly after
# triggered-submitted and then we have inavlid
# value in `reqids2txids` sent over ws.send()??
log.error('TOO FAST CANCEL/EDIT') log.error('TOO FAST CANCEL/EDIT')
reqids2txids[reqid] = TooFastEdit(reqid) # reqids2txids[reqid] = TooFastEdit(reqid)
toofastedit.add(reqid)
reqids2txids[reqid] = reqid
await ems_order_stream.send( await ems_order_stream.send(
BrokerdError( BrokerdError(
oid=msg['oid'], oid=msg['oid'],
@ -199,7 +206,10 @@ async def handle_order_requests(
# XXX: not sure if this block ever gets hit now? # XXX: not sure if this block ever gets hit now?
log.error('TOO FAST EDIT') log.error('TOO FAST EDIT')
reqids2txids[reqid] = TooFastEdit(reqid) # reqids2txids[reqid] = TooFastEdit(reqid)
reqids2txids[reqid] = reqid
toofastedit.add(reqid)
await tractor.pause()
await ems_order_stream.send( await ems_order_stream.send(
BrokerdError( BrokerdError(
oid=msg['oid'], oid=msg['oid'],
@ -665,6 +675,10 @@ async def open_trade_dialog(
token: str = await client.get_ws_token() token: str = await client.get_ws_token()
# XXX tracks EMS orders which are updated too quickly
# on the emds side with sync-issues on the kraken side.
toofastedit: set[int] = set()
ws: NoBsWs ws: NoBsWs
async with ( async with (
ctx.open_stream() as ems_stream, ctx.open_stream() as ems_stream,
@ -680,15 +694,16 @@ async def open_trade_dialog(
trio.open_nursery() as tn, trio.open_nursery() as tn,
): ):
# task for processing inbound requests from ems # task for processing inbound requests from ems
tn.start_soon( tn.start_soon(partial(
handle_order_requests, handle_order_requests,
ws, ws=ws,
client, client=client,
ems_stream, ems_stream=ems_stream,
apiflows, apiflows=apiflows,
ids, ids=ids,
reqids2txids, reqids2txids=reqids2txids,
) toofastedit=toofastedit,
))
# enter relay loop # enter relay loop
await handle_order_updates( await handle_order_updates(
@ -699,6 +714,7 @@ async def open_trade_dialog(
apiflows=apiflows, apiflows=apiflows,
ids=ids, ids=ids,
reqids2txids=reqids2txids, reqids2txids=reqids2txids,
toofastedit=toofastedit,
acnt=acnt, acnt=acnt,
ledger=ledger, ledger=ledger,
acctid=acctid, acctid=acctid,
@ -714,6 +730,7 @@ async def handle_order_updates(
apiflows: OrderDialogs, apiflows: OrderDialogs,
ids: bidict[str, int], ids: bidict[str, int],
reqids2txids: bidict[int, str], reqids2txids: bidict[int, str],
toofastedit: set[int],
acnt: Account, acnt: Account,
# transaction records which will be updated # transaction records which will be updated
@ -1021,10 +1038,12 @@ async def handle_order_updates(
# <-> ems dialog. # <-> ems dialog.
if ( if (
status == 'open' status == 'open'
and isinstance( and
reqids2txids.get(reqid), reqid in toofastedit
TooFastEdit # isinstance(
) # reqids2txids.get(reqid),
# TooFastEdit
# )
): ):
# TODO: don't even allow this case # TODO: don't even allow this case
# by not moving the client side line # by not moving the client side line
@ -1186,7 +1205,9 @@ async def handle_order_updates(
txid txid
# we throttle too-fast-requests on the ems side # we throttle too-fast-requests on the ems side
and not isinstance(txid, TooFastEdit) and
reqid in toofastedit
# not isinstance(txid, TooFastEdit)
): ):
# client was editting too quickly # client was editting too quickly
# so we instead cancel this order # so we instead cancel this order