`.storage.__init__`: code styling updates

For `[ib]` adjust content to match changes to the
`dockering/ib/README.rst` and for `[deribit]` toss in the WIP options
related params for anyone who wants to play around with @nt's work.

config/brokers.toml

@@ -1,6 +1,5 @@
-################
 # ---- CEXY ----
-################
+
 [binance]
 accounts.paper = 'paper'
 
@@ -13,28 +12,41 @@ accounts.spot = 'spot'
 spot.use_testnet = false
 spot.api_key = ''
 spot.api_secret = ''
+# ------ binance ------
 
 
 [deribit]
+# std assets
 key_id = ''
 key_secret = ''
+# options
+accounts.option = 'option'
+option.use_testnet = false
+option.key_id = ''
+option.key_secret = ''
+# aux logging from `cryptofeed`
+option.log.filename = 'cryptofeed.log'
+option.log.level = 'DEBUG'
+option.log.disabled = true
+# ------ deribit ------
 
 
 [kraken]
 key_descr = ''
 api_key = ''
 secret = ''
+# ------ kraken ------
 
 
 [kucoin]
 key_id = ''
 key_secret = ''
 key_passphrase = ''
+# ------ kucoin ------
 
 
-################
 # -- BROKERZ ---
-################
+
 [questrade]
 refresh_token = ''
 access_token = ''
@@ -42,44 +54,55 @@ api_server = 'https://api06.iq.questrade.com/'
 expires_in = 1800
 token_type = 'Bearer'
 expires_at = 1616095326.355846
+# ------ questrade ------
 
 
 [ib]
+# define the (set of) host-port socketaddrs that
+# brokerd.ib will scan to connect to an API endpoint
+# (ib-gw or ib-tws listening instances)
 hosts = [
     '127.0.0.1',
 ]
-# XXX: the order in which ports will be scanned
-# (by the `brokerd` daemon-actor)
-# is determined # by the line order here.
-# TODO: when we eventually spawn gateways in our
-# container, we can just dynamically allocate these
-# using IBC.
 ports = [
     4002,  # gw
     7497,  # tws
 ]
 
-# XXX: for a paper account the flex web query service
+# When API endpoints are being scanned durin startup, the order
-# is not supported so you have to manually download
+# of user-defined-account "names" (as defined below) here
-# and XML report and put it in a location that can be
+# determines which py-client connection is given priority to be
-# accessed by the ``brokerd.ib`` backend code for parsing.
+# used for data-feed-requests by according to whichever client
-flex_token = ''
+# connected to an API endpoing which reported the equivalent
-flex_trades_query_id = ''  # live account
+# account number for that name.
-
-# when clients are being scanned this determines
-# which clients are preferred to be used for data
-# feeds based on the order of account names, if
-# detected as active on an API client.
 prefer_data_account = [
     'paper',
     'margin',
     'ira',
 ]
 
+# For long-term trades txn (transaction) history
+# processing (i.e your txn ledger with IB) you can
+# (automatically for live accounts) query the FLEX
+# report system for past history.
+#
+# (For paper accounts the web query service
+# is not supported so you have to manually download
+# an XML report and put it in a location that can be
+# accessed by our `brokerd.ib` backend code for parsing).
+#
+flex_token = ''
+flex_trades_query_id = ''  # live account
+
+# define "aliases" (names) for each account number
+# such that the names can be reffed and logged throughout
+# `piker.accounting` subsys and more easily
+# referred to  by the user.
+#
+# These keys will be the set exposed through the order-mode
+# account-selection UI so that numbers are never shown.
 [ib.accounts]
-# the order in which accounts will be selectable
+paper = 'DU0000000'  # <- literal account #
-# in the order mode UI (if found via clients during
+margin = 'U0000000'
-# API-app scanning)when a new symbol is loaded.
+ira = 'U0000000'
-paper = 'XX0000000'
+# ------ ib ------
-margin = 'X0000000'
-ira = 'X0000000'

examples/basic_order_bot.py

@@ -121,6 +121,7 @@ async def bot_main():
             # tick_throttle=10,
         ) as feed,
 
+        tractor.trionics.collapse_eg(),
         trio.open_nursery() as tn,
     ):
         assert accounts

piker/accounting/__init__.py

@@ -33,7 +33,6 @@ from ._pos import (
     Account,
     load_account,
     load_account_from_ledger,
-    open_pps,
     open_account,
     Position,
 )
@@ -68,7 +67,6 @@ __all__ = [
     'load_account_from_ledger',
     'mk_allocator',
     'open_account',
-    'open_pps',
     'open_trade_ledger',
     'unpack_fqme',
     'DerivTypes',

piker/accounting/_pos.py

@@ -356,17 +356,20 @@ class Position(Struct):
     ) -> bool:
         '''
         Update clearing table by calculating the rolling ppu and
-        (accumulative) size in both the clears entry and local
+        (accumulative) size in both the clears entry and local attrs
-        attrs state.
+        state.
 
         Inserts are always done in datetime sorted order.
 
         '''
-        # added: bool = False
         tid: str = t.tid
         if tid in self._events:
-            log.warning(f'{t} is already added?!')
+            log.debug(
-            # return added
+                f'Txn is already added?\n'
+                f'\n'
+                f'{t}\n'
+            )
+            return False
 
         # TODO: apparently this IS possible with a dict but not
         # common and probably not that beneficial unless we're also
@@ -447,6 +450,12 @@ class Position(Struct):
     # def suggest_split(self) -> float:
     #     ...
 
+    # ?TODO, for sending rendered state over the wire?
+    # def summary(self) -> PositionSummary:
+        # do minimal conversion to a subset of fields
+        # currently defined in `.clearing._messages.BrokerdPosition`
+
+
 
 class Account(Struct):
     '''
@@ -490,9 +499,9 @@ class Account(Struct):
 
     def update_from_ledger(
         self,
-        ledger: TransactionLedger | dict[str, Transaction],
+        ledger: TransactionLedger|dict[str, Transaction],
         cost_scalar: float = 2,
-        symcache: SymbologyCache | None = None,
+        symcache: SymbologyCache|None = None,
 
         _mktmap_table: dict[str, MktPair] | None = None,
 
@@ -731,7 +740,7 @@ class Account(Struct):
                     else:
                         # TODO: we reallly need a diff set of
                         # loglevels/colors per subsys.
-                        log.warning(
+                        log.debug(
                             f'Recent position for {fqme} was closed!'
                         )
 
@@ -745,7 +754,7 @@ class Account(Struct):
         # XXX WTF: if we use a tomlkit.Integer here we get this
         # super weird --1 thing going on for cumsize!?1!
         # NOTE: the fix was to always float() the size value loaded
-        # in open_pps() below!
+        # in open_account() below!
         config.write(
             config=self.conf,
             path=self.conf_path,
@@ -929,7 +938,6 @@ def open_account(
             clears_table['dt'] = dt
             trans.append(Transaction(
                 fqme=bs_mktid,
-                # sym=mkt,
                 bs_mktid=bs_mktid,
                 tid=tid,
                 # XXX: not sure why sometimes these are loaded as
@@ -952,7 +960,18 @@ def open_account(
         ):
             expiry: pendulum.DateTime = pendulum.parse(expiry)
 
-        pp = pp_objs[bs_mktid] = Position(
+        # !XXX, should never be duplicates over
+        # a backend-(broker)-system's unique market-IDs!
+        if pos := pp_objs.get(bs_mktid):
+            if mkt != pos.mkt:
+                log.warning(
+                    f'Duplicated position but diff `MktPair.fqme` ??\n'
+                    f'bs_mktid: {bs_mktid!r}\n'
+                    f'pos.mkt: {pos.mkt}\n'
+                    f'mkt: {mkt}\n'
+                )
+        else:
+            pos = pp_objs[bs_mktid] = Position(
                 mkt,
                 split_ratio=split_ratio,
                 bs_mktid=bs_mktid,
@@ -964,8 +983,13 @@ def open_account(
         # state, since today's records may have already been
         # processed!
         for t in trans:
-            pp.add_clear(t)
+            added: bool = pos.add_clear(t)
-
+            if not added:
+                log.warning(
+                    f'Txn already recorded in pp ??\n'
+                    f'\n'
+                    f'{t}\n'
+                )
     try:
         yield acnt
     finally:
@@ -973,20 +997,6 @@ def open_account(
             acnt.write_config()
 
 
-# TODO: drop the old name and THIS!
-@cm
-def open_pps(
-    *args,
-    **kwargs,
-) -> Generator[Account, None, None]:
-    log.warning(
-        '`open_pps()` is now deprecated!\n'
-        'Please use `with open_account() as cnt:`'
-    )
-    with open_account(*args, **kwargs) as acnt:
-        yield acnt
-
-
 def load_account_from_ledger(
 
     brokername: str,

piker/accounting/calc.py

@@ -22,7 +22,9 @@ you know when you're losing money (if possible) XD
 from __future__ import annotations
 from collections.abc import ValuesView
 from contextlib import contextmanager as cm
+from functools import partial
 from math import copysign
+from pprint import pformat
 from typing import (
     Any,
     Callable,
@@ -30,6 +32,7 @@ from typing import (
     TYPE_CHECKING,
 )
 
+from tractor.devx import maybe_open_crash_handler
 import polars as pl
 from pendulum import (
     DateTime,
@@ -37,12 +40,16 @@ from pendulum import (
     parse,
 )
 
+from ..log import get_logger
+
 if TYPE_CHECKING:
     from ._ledger import (
         Transaction,
         TransactionLedger,
     )
 
+log = get_logger(__name__)
+
 
 def ppu(
     clears: Iterator[Transaction],
@@ -238,6 +245,9 @@ def iter_by_dt(
 
     def dyn_parse_to_dt(
         tx: tuple[str, dict[str, Any]] | Transaction,
+
+        debug: bool = False,
+        _invalid: list|None = None,
     ) -> DateTime:
 
         # handle `.items()` inputs
@@ -250,52 +260,90 @@ def iter_by_dt(
         # get best parser for this record..
         for k in parsers:
             if (
-                isdict and k in tx
+                (v := getattr(tx, k, None))
                 or
-                getattr(tx, k, None)
+                (
+                    isdict
+                    and
+                    (v := tx.get(k))
+                )
             ):
-                v = (
-                    tx[k] if isdict
-                    else tx.dt
-                )
-                assert v is not None, (
-                    f'No valid value for `{k}`!?'
-                )
-
                 # only call parser on the value if not None from
                 # the `parsers` table above (when NOT using
                 # `.get()`), otherwise pass through the value and
                 # sort on it directly
                 if (
                     not isinstance(v, DateTime)
-                    and (parser := parsers.get(k))
+                    and
+                    (parser := parsers.get(k))
                 ):
-                    return parser(v)
+                    ret = parser(v)
                 else:
-                    return v
+                    ret = v
+
+                return ret
 
             else:
-            # TODO: move to top?
+                log.debug(
-            from piker.log import get_logger
+                    f'Parser-field not found in txn\n'
-            log = get_logger(__name__)
+                    f'\n'
+                    f'parser-field: {k!r}\n'
+                    f'txn: {tx!r}\n'
+                    f'\n'
+                    f'Trying next..\n'
+                )
+                continue
 
+        # XXX: we should never really get here bc it means some kinda
+        # bad txn-record (field) data..
+        #
+        # -> set the `debug_mode = True` if you want to trace such
+        # cases from REPL ;)
+        else:
             # XXX: we should really never get here..
             # only if a ledger record has no expected sort(able)
             # field will we likely hit this.. like with ze IB.
             # if no sortable field just deliver epoch?
             log.warning(
                 'No (time) sortable field for TXN:\n'
-                f'{tx}\n'
+                f'{tx!r}\n'
             )
-            return from_timestamp(0)
+            report: str = (
-            # breakpoint()
+                f'No supported time-field found in txn !?\n'
+                f'\n'
+                f'supported-time-fields: {parsers!r}\n'
+                f'\n'
+                f'txn: {tx!r}\n'
+            )
+            if debug:
+                with maybe_open_crash_handler(
+                    pdb=debug,
+                    raise_on_exit=False,
+                ):
+                    raise ValueError(report)
+            else:
+                log.error(report)
 
+            if _invalid is not None:
+                _invalid.append(tx)
+                return from_timestamp(0.)
 
-    entry: tuple[str, dict] | Transaction
+    entry: tuple[str, dict]|Transaction
+    invalid: list = []
     for entry in sorted(
         records,
-        key=key or dyn_parse_to_dt,
+        key=key or partial(
+            dyn_parse_to_dt,
+            _invalid=invalid,
+        ),
     ):
+        if entry in invalid:
+            log.warning(
+                f'Ignoring txn w invalid timestamp ??\n'
+                f'{pformat(entry)}\n'
+            )
+            continue
+
         # NOTE the type sig above; either pairs or txns B)
         yield entry
 
@@ -358,6 +406,7 @@ def open_ledger_dfs(
     acctname: str,
 
     ledger: TransactionLedger | None = None,
+    debug_mode: bool = False,
 
     **kwargs,
 
@@ -372,8 +421,10 @@ def open_ledger_dfs(
     can update the ledger on exit.
 
     '''
-    from piker.toolz import open_crash_handler
+    with maybe_open_crash_handler(
-    with open_crash_handler():
+        pdb=debug_mode,
+        # raise_on_exit=False,
+    ):
         if not ledger:
             import time
             from ._ledger import open_trade_ledger

piker/brokers/_daemon.py

@@ -96,7 +96,10 @@ async def _setup_persistent_brokerd(
     # - `open_symbol_search()`
     # NOTE: see ep invocation details inside `.data.feed`.
     try:
-        async with trio.open_nursery() as service_nursery:
+        async with (
+            tractor.trionics.collapse_eg(),
+            trio.open_nursery() as service_nursery
+        ):
             bus: _FeedsBus = feed.get_feed_bus(
                 brokername,
                 service_nursery,

piker/brokers/binance/broker.py

@@ -440,6 +440,7 @@ async def open_trade_dialog(
             #   - ledger: TransactionLedger
 
             async with (
+                tractor.trionics.collapse_eg(),
                 trio.open_nursery() as tn,
                 ctx.open_stream() as ems_stream,
             ):

piker/brokers/binance/feed.py

@@ -448,7 +448,6 @@ async def subscribe(
 
 
 async def stream_quotes(
-
     send_chan: trio.abc.SendChannel,
     symbols: list[str],
     feed_is_live: trio.Event,
@@ -460,6 +459,7 @@ async def stream_quotes(
 ) -> None:
 
     async with (
+        tractor.trionics.maybe_raise_from_masking_exc(),
         send_chan as send_chan,
         open_cached_client('binance') as client,
     ):

piker/brokers/deribit/api.py

@@ -31,7 +31,7 @@ from typing import (
     Callable,
 )
 
-import pendulum
+from pendulum import now
 import trio
 from trio_typing import TaskStatus
 from rapidfuzz import process as fuzzy
@@ -39,6 +39,7 @@ import numpy as np
 from tractor.trionics import (
     broadcast_receiver,
     maybe_open_context
+    collapse_eg,
 )
 from tractor import to_asyncio
 # XXX WOOPS XD
@@ -432,6 +433,7 @@ async def get_client(
 ) -> Client:
 
     async with (
+        collapse_eg(),
         trio.open_nursery() as n,
         open_jsonrpc_session(
             _testnet_ws_url, dtype=JSONRPCResult) as json_rpc

piker/brokers/ib/api.py

@@ -48,6 +48,7 @@ from bidict import bidict
 import trio
 import tractor
 from tractor import to_asyncio
+from tractor import trionics
 from pendulum import (
     from_timestamp,
     DateTime,
@@ -1369,8 +1370,8 @@ async def load_clients_for_trio(
     '''
     Pure async mngr proxy to ``load_aio_clients()``.
 
-    This is a bootstrap entrypoing to call from
+    This is a bootstrap entrypoint to call from
-    a ``tractor.to_asyncio.open_channel_from()``.
+    a `tractor.to_asyncio.open_channel_from()`.
 
     '''
     async with load_aio_clients(
@@ -1391,7 +1392,10 @@ async def open_client_proxies() -> tuple[
     async with (
         tractor.trionics.maybe_open_context(
             acm_func=tractor.to_asyncio.open_channel_from,
-            kwargs={'target': load_clients_for_trio},
+            kwargs={
+                'target': load_clients_for_trio,
+                # ^XXX, kwarg to `open_channel_from()`
+            },
 
             # lock around current actor task access
             # TODO: maybe this should be the default in tractor?
@@ -1584,7 +1588,8 @@ async def open_client_proxy(
             event_consumers=event_table,
         ) as (first, chan),
 
-        trio.open_nursery() as relay_n,
+        trionics.collapse_eg(),  # loose-ify
+        trio.open_nursery() as relay_tn,
     ):
 
         assert isinstance(first, Client)
@@ -1624,7 +1629,7 @@ async def open_client_proxy(
 
                     continue
 
-        relay_n.start_soon(relay_events)
+        relay_tn.start_soon(relay_events)
 
         yield proxy
 

piker/brokers/ib/broker.py

@@ -34,6 +34,7 @@ import trio
 from trio_typing import TaskStatus
 import tractor
 from tractor.to_asyncio import LinkedTaskChannel
+from tractor import trionics
 from ib_insync.contract import (
     Contract,
 )
@@ -407,7 +408,7 @@ async def update_and_audit_pos_msg(
 
     # TODO: make this a "propaganda" log level?
     if ibpos.avgCost != msg.avg_price:
-        log.warning(
+        log.debug(
             f'IB "FIFO" avg price for {msg.symbol} is DIFF:\n'
             f'ib: {ibfmtmsg}\n'
             '---------------------------\n'
@@ -738,7 +739,7 @@ async def open_trade_dialog(
                                 f'UNEXPECTED POSITION says IB => {msg.symbol}\n'
                                 'Maybe they LIQUIDATED YOU or your ledger is wrong?\n'
                             )
-                            log.error(logmsg)
+                            log.debug(logmsg)
 
             await ctx.started((
                 all_positions,
@@ -747,21 +748,22 @@ async def open_trade_dialog(
 
             async with (
                 ctx.open_stream() as ems_stream,
-                trio.open_nursery() as n,
+                trionics.collapse_eg(),
+                trio.open_nursery() as tn,
             ):
                 # relay existing open orders to ems
                 for msg in order_msgs:
                     await ems_stream.send(msg)
 
                 for client in set(aioclients.values()):
-                    trade_event_stream: LinkedTaskChannel = await n.start(
+                    trade_event_stream: LinkedTaskChannel = await tn.start(
                         open_trade_event_stream,
                         client,
                     )
 
                     # start order request handler **before** local trades
                     # event loop
-                    n.start_soon(
+                    tn.start_soon(
                         handle_order_requests,
                         ems_stream,
                         accounts_def,
@@ -769,7 +771,7 @@ async def open_trade_dialog(
                     )
 
                     # allocate event relay tasks for each client connection
-                    n.start_soon(
+                    tn.start_soon(
                         deliver_trade_events,
 
                         trade_event_stream,

piker/clearing/_client.py

@@ -25,7 +25,10 @@ from typing import TYPE_CHECKING
 
 import trio
 import tractor
-from tractor.trionics import broadcast_receiver
+from tractor.trionics import (
+    broadcast_receiver,
+    collapse_eg,
+)
 
 from ._util import (
     log,  # sub-sys logger
@@ -285,8 +288,11 @@ async def open_ems(
             client._ems_stream = trades_stream
 
             # start sync code order msg delivery task
-            async with trio.open_nursery() as n:
+            async with (
-                n.start_soon(
+                collapse_eg(),
+                trio.open_nursery() as tn,
+            ):
+                tn.start_soon(
                     relay_orders_from_sync_code,
                     client,
                     fqme,
@@ -302,4 +308,4 @@ async def open_ems(
                 )
 
                 # stop the sync-msg-relay task on exit.
-                n.cancel_scope.cancel()
+                tn.cancel_scope.cancel()

piker/clearing/_ems.py

@@ -42,6 +42,7 @@ from bidict import bidict
 import trio
 from trio_typing import TaskStatus
 import tractor
+from tractor import trionics
 
 from ._util import (
     log,  # sub-sys logger
@@ -161,7 +162,7 @@ async def clear_dark_triggers(
 
     router: Router,
     brokerd_orders_stream: tractor.MsgStream,
-    quote_stream: tractor.ReceiveMsgStream,  # noqa
+    quote_stream: tractor.MsgStream,
     broker: str,
     fqme: str,
 
@@ -177,6 +178,7 @@ async def clear_dark_triggers(
     '''
     # XXX: optimize this for speed!
     # TODO:
+    # - port to the new ringbuf stuff in `tractor.ipc`!
     # - numba all this!
     # - this stream may eventually contain multiple symbols
     quote_stream._raise_on_lag = False
@@ -386,6 +388,7 @@ async def open_brokerd_dialog(
     for ep_name in [
         'open_trade_dialog',  # probably final name?
         'trades_dialogue',  # legacy
+        # ^!TODO, rm this since all backends ported no ?!?
     ]:
         trades_endpoint = getattr(
             brokermod,
@@ -499,7 +502,7 @@ class Router(Struct):
 
     '''
     # setup at actor spawn time
-    nursery: trio.Nursery
+    _tn: trio.Nursery
 
     # broker to book map
     books: dict[str, DarkBook] = {}
@@ -669,7 +672,7 @@ class Router(Struct):
                 # dark book clearing loop, also lives with parent
                 # daemon to allow dark order clearing while no
                 # client is connected.
-                self.nursery.start_soon(
+                self._tn.start_soon(
                     clear_dark_triggers,
                     self,
                     relay.brokerd_stream,
@@ -692,7 +695,7 @@ class Router(Struct):
 
                 # spawn a ``brokerd`` order control dialog stream
                 # that syncs lifetime with the parent `emsd` daemon.
-                self.nursery.start_soon(
+                self._tn.start_soon(
                     translate_and_relay_brokerd_events,
                     broker,
                     relay.brokerd_stream,
@@ -766,10 +769,12 @@ async def _setup_persistent_emsd(
 
     global _router
 
-    # open a root "service nursery" for the ``emsd`` actor
+    # open a root "service task-nursery" for the `emsd`-actor
-    async with trio.open_nursery() as service_nursery:
+    async with (
-
+        trionics.collapse_eg(),
-        _router = Router(nursery=service_nursery)
+        trio.open_nursery() as tn
+    ):
+        _router = Router(_tn=tn)
 
         # TODO: send back the full set of persistent
         # orders/execs?
@@ -1023,8 +1028,18 @@ async def translate_and_relay_brokerd_events(
                     )
 
                 if status == 'closed':
-                    log.info(f'Execution for {oid} is complete!')
+                    log.info(
-                    status_msg = book._active.pop(oid)
+                        f'Execution is complete!\n'
+                        f'oid: {oid!r}\n'
+                    )
+                    status_msg = book._active.pop(oid, None)
+                    if status_msg is None:
+                        log.warning(
+                            f'Order was already cleared from book ??\n'
+                            f'oid: {oid!r}\n'
+                            f'\n'
+                            f'Maybe the order cancelled before submitted ??\n'
+                        )
 
                 elif status == 'canceled':
                     log.cancel(f'Cancellation for {oid} is complete!')
@@ -1518,7 +1533,7 @@ async def maybe_open_trade_relays(
         loglevel: str = 'info',
     ):
 
-        fqme, relay, feed, client_ready = await _router.nursery.start(
+        fqme, relay, feed, client_ready = await _router._tn.start(
             _router.open_trade_relays,
             fqme,
             exec_mode,
@@ -1548,19 +1563,18 @@ async def maybe_open_trade_relays(
 
 @tractor.context
 async def _emsd_main(
-    ctx: tractor.Context,
+    ctx: tractor.Context,   # becomes `ems_ctx` below
     fqme: str,
     exec_mode: str,  # ('paper', 'live')
     loglevel: str|None = None,
 
-) -> tuple[
+) -> tuple[  # `ctx.started()` value!
-    dict[
+    dict[  # positions
-        # brokername, acctid
+        tuple[str, str],  # brokername, acctid
-        tuple[str, str],
         list[BrokerdPosition],
     ],
-    list[str],
+    list[str],  # accounts
-    dict[str, Status],
+    dict[str, Status],  # dialogs
 ]:
     '''
     EMS (sub)actor entrypoint providing the execution management

piker/clearing/_messages.py

@@ -301,6 +301,9 @@ class BrokerdError(Struct):
 
 # TODO: yeah, so we REALLY need to completely deprecate
 # this and use the `.accounting.Position` msg-type instead..
+# -[ ] an alternative might be to add a `Position.summary() ->
+#   `PositionSummary`-msg that we generate since `Position` has a lot
+#   of fields by default we likely don't want to send over the wire?
 class BrokerdPosition(Struct):
     '''
     Position update event from brokerd.
@@ -313,3 +316,4 @@ class BrokerdPosition(Struct):
     avg_price: float
     currency: str = ''
     name: str = 'position'
+    bs_mktid: str|int|None = None

piker/cli/__init__.py

@@ -134,8 +134,8 @@ def pikerd(
     Spawn the piker broker-daemon.
 
     '''
-    from tractor.devx import maybe_open_crash_handler
+    # from tractor.devx import maybe_open_crash_handler
-    with maybe_open_crash_handler(pdb=pdb):
+    # with maybe_open_crash_handler(pdb=False):
     log = get_console_log(loglevel, name='cli')
 
     if pdb:
@@ -178,16 +178,14 @@ def pikerd(
 
     async def main():
         service_mngr: service.Services
-
         async with (
             service.open_pikerd(
                 registry_addrs=regaddrs,
                 loglevel=loglevel,
                 debug_mode=pdb,
-
+                # enable_transports=['uds'],
-                ) as service_mngr,  # normally delivers a ``Services`` handle
+                enable_transports=['tcp'],
-
+            ) as service_mngr,
-                # AsyncExitStack() as stack,
         ):
             assert service_mngr
             # ?TODO? spawn all other sub-actor daemons according to
@@ -309,6 +307,10 @@ def services(config, tl, ports):
     if not ports:
         ports = [_default_registry_port]
 
+    addr = tractor._addr.wrap_address(
+        addr=(host, ports[0])
+    )
+
     async def list_services():
         nonlocal host
         async with (
@@ -316,16 +318,18 @@ def services(config, tl, ports):
                 name='service_query',
                 loglevel=config['loglevel'] if tl else None,
             ),
-            tractor.get_arbiter(
+            tractor.get_registry(
-                host=host,
+                addr=addr,
-                port=ports[0]
             ) as portal
         ):
-            registry = await portal.run_from_ns('self', 'get_registry')
+            registry = await portal.run_from_ns(
+                'self',
+                'get_registry',
+            )
             json_d = {}
             for key, socket in registry.items():
-                host, port = socket
+                json_d[key] = f'{socket}'
-                json_d[key] = f'{host}:{port}'
+
             click.echo(f"{colorize_json(json_d)}")
 
     trio.run(list_services)

piker/data/_web_bs.py

@@ -27,7 +27,6 @@ from functools import partial
 from types import ModuleType
 from typing import (
     Any,
-    Optional,
     Callable,
     AsyncContextManager,
     AsyncGenerator,
@@ -35,6 +34,7 @@ from typing import (
 )
 import json
 
+import tractor
 import trio
 from trio_typing import TaskStatus
 from trio_websocket import (
@@ -167,7 +167,7 @@ async def _reconnect_forever(
 
     async def proxy_msgs(
         ws: WebSocketConnection,
-        pcs: trio.CancelScope,  # parent cancel scope
+        rent_cs: trio.CancelScope,  # parent cancel scope
     ):
         '''
         Receive (under `timeout` deadline) all msgs from from underlying
@@ -192,7 +192,7 @@ async def _reconnect_forever(
                         f'{url} connection bail with:'
                     )
                     await trio.sleep(0.5)
-                    pcs.cancel()
+                    rent_cs.cancel()
 
                     # go back to reonnect loop in parent task
                     return
@@ -204,7 +204,7 @@ async def _reconnect_forever(
                         f'{src_mod}\n'
                         'WS feed seems down and slow af.. reconnecting\n'
                     )
-                    pcs.cancel()
+                    rent_cs.cancel()
 
                     # go back to reonnect loop in parent task
                     return
@@ -228,7 +228,12 @@ async def _reconnect_forever(
     nobsws._connected = trio.Event()
     task_status.started()
 
-    while not snd._closed:
+    mc_state: trio._channel.MemoryChannelState = snd._state
+    while (
+        mc_state.open_receive_channels > 0
+        and
+        mc_state.open_send_channels > 0
+    ):
         log.info(
             f'{src_mod}\n'
             f'{url} trying (RE)CONNECT'
@@ -237,10 +242,11 @@ async def _reconnect_forever(
         ws: WebSocketConnection
         try:
             async with (
-                trio.open_nursery() as n,
                 open_websocket_url(url) as ws,
+                tractor.trionics.collapse_eg(),
+                trio.open_nursery() as tn,
             ):
-                cs = nobsws._cs = n.cancel_scope
+                cs = nobsws._cs = tn.cancel_scope
                 nobsws._ws = ws
                 log.info(
                     f'{src_mod}\n'
@@ -248,7 +254,7 @@ async def _reconnect_forever(
                 )
 
                 # begin relay loop to forward msgs
-                n.start_soon(
+                tn.start_soon(
                     proxy_msgs,
                     ws,
                     cs,
@@ -262,7 +268,7 @@ async def _reconnect_forever(
 
                     # TODO: should we return an explicit sub-cs
                     # from this fixture task?
-                    await n.start(
+                    await tn.start(
                         open_fixture,
                         fixture,
                         nobsws,
@@ -272,11 +278,23 @@ async def _reconnect_forever(
                 # to let tasks run **inside** the ws open block above.
                 nobsws._connected.set()
                 await trio.sleep_forever()
-        except HandshakeError:
+
+        except (
+            HandshakeError,
+            ConnectionRejected,
+        ):
             log.exception('Retrying connection')
+            await trio.sleep(0.5)  # throttle
 
-        # ws & nursery block ends
+        except BaseException as _berr:
+            berr = _berr
+            log.exception(
+                'Reconnect-attempt failed ??\n'
+            )
+            await trio.sleep(0.2)  # throttle
+            raise berr
 
+        #|_ws & nursery block ends
         nobsws._connected = trio.Event()
         if cs.cancelled_caught:
             log.cancel(
@@ -324,21 +342,25 @@ async def open_autorecon_ws(
     connetivity errors, or some user defined recv timeout.
 
     You can provide a ``fixture`` async-context-manager which will be
-    entered/exitted around each connection reset; eg. for (re)requesting
+    entered/exitted around each connection reset; eg. for
-    subscriptions without requiring streaming setup code to rerun.
+    (re)requesting subscriptions without requiring streaming setup
+    code to rerun.
 
     '''
     snd: trio.MemorySendChannel
     rcv: trio.MemoryReceiveChannel
     snd, rcv = trio.open_memory_channel(616)
 
-    async with trio.open_nursery() as n:
+    async with (
+        tractor.trionics.collapse_eg(),
+        trio.open_nursery() as tn
+    ):
         nobsws = NoBsWs(
             url,
             rcv,
             msg_recv_timeout=msg_recv_timeout,
         )
-        await n.start(
+        await tn.start(
             partial(
                 _reconnect_forever,
                 url,
@@ -351,11 +373,10 @@ async def open_autorecon_ws(
         await nobsws._connected.wait()
         assert nobsws._cs
         assert nobsws.connected()
-
         try:
             yield nobsws
         finally:
-            n.cancel_scope.cancel()
+            tn.cancel_scope.cancel()
 
 
 '''
@@ -368,8 +389,8 @@ of msgs over a `NoBsWs`.
 class JSONRPCResult(Struct):
     id: int
     jsonrpc: str = '2.0'
-    result: Optional[dict] = None
+    result: dict|None = None
-    error: Optional[dict] = None
+    error: dict|None = None
 
 
 @acm

piker/data/feed.py

@@ -39,6 +39,7 @@ from typing import (
     AsyncContextManager,
     Awaitable,
     Sequence,
+    TYPE_CHECKING,
 )
 
 import trio
@@ -75,6 +76,10 @@ from ._sampling import (
     uniform_rate_send,
 )
 
+if TYPE_CHECKING:
+    from tractor._addr import Address
+    from tractor.msg.types import Aid
+
 
 class Sub(Struct, frozen=True):
     '''
@@ -725,7 +730,10 @@ class Feed(Struct):
                 async for msg in stream:
                     await tx.send(msg)
 
-        async with trio.open_nursery() as nurse:
+        async with (
+            tractor.trionics.collapse_eg(),
+            trio.open_nursery() as nurse
+        ):
             # spawn a relay task for each stream so that they all
             # multiplex to a common channel.
             for brokername in mods:
@@ -899,19 +907,19 @@ async def open_feed(
             feed.portals[brokermod] = portal
 
             # fill out "status info" that the UI can show
-            host, port = portal.channel.raddr
+            chan: tractor.Channel = portal.chan
-            if host == '127.0.0.1':
+            raddr: Address = chan.raddr
-                host = 'localhost'
+            aid: Aid = chan.aid
-
+            # TAG_feed_status_update
             feed.status.update({
-                'actor_name': portal.channel.uid[0],
+                'actor_id': aid,
-                'host': host,
+                'actor_short_id': f'{aid.name}@{aid.pid}',
-                'port': port,
+                'ipc': chan.raddr.proto_key,
+                'ipc_addr': raddr,
                 'hist_shm': 'NA',
                 'rt_shm': 'NA',
-                'throttle_rate': tick_throttle,
+                'throttle_hz': tick_throttle,
             })
-            # feed.status.update(init_msg.pop('status', {}))
 
             # (allocate and) connect to any feed bus for this broker
             bus_ctxs.append(

piker/fsp/_engine.py

@@ -498,6 +498,7 @@ async def cascade(
 
         func_name: str = func.__name__
         async with (
+            tractor.trionics.collapse_eg(),  # avoid multi-taskc tb in console
             trio.open_nursery() as tn,
         ):
             # TODO: might be better to just make a "restart" method where

piker/service/_actor_runtime.py

@@ -107,17 +107,22 @@ async def open_piker_runtime(
         async with (
             tractor.open_root_actor(
 
-                # passed through to ``open_root_actor``
+                # passed through to `open_root_actor`
                 registry_addrs=registry_addrs,
                 name=name,
+                start_method=start_method,
                 loglevel=loglevel,
                 debug_mode=debug_mode,
-                start_method=start_method,
+
+                # XXX NOTE MEMBER DAT der's a perf hit yo!!
+                # https://greenback.readthedocs.io/en/latest/principle.html#performance
+                maybe_enable_greenback=True,
 
                 # TODO: eventually we should be able to avoid
                 # having the root have more then permissions to
                 # spawn other specialized daemons I think?
                 enable_modules=enable_modules,
+                hide_tb=False,
 
                 **tractor_kwargs,
             ) as actor,
@@ -200,7 +205,8 @@ async def open_pikerd(
             reg_addrs,
         ),
         tractor.open_nursery() as actor_nursery,
-        trio.open_nursery() as service_nursery,
+        tractor.trionics.collapse_eg(),
+        trio.open_nursery() as service_tn,
     ):
         for addr in reg_addrs:
             if addr not in root_actor.accept_addrs:
@@ -211,7 +217,7 @@ async def open_pikerd(
 
         # assign globally for future daemon/task creation
         Services.actor_n = actor_nursery
-        Services.service_n = service_nursery
+        Services.service_n = service_tn
         Services.debug_mode = debug_mode
 
         try:
@@ -221,7 +227,7 @@ async def open_pikerd(
             # TODO: is this more clever/efficient?
             # if 'samplerd' in Services.service_tasks:
             #     await Services.cancel_service('samplerd')
-            service_nursery.cancel_scope.cancel()
+            service_tn.cancel_scope.cancel()
 
 
 # TODO: do we even need this?
@@ -256,7 +262,10 @@ async def maybe_open_pikerd(
     loglevel: str | None = None,
     **kwargs,
 
-) -> tractor._portal.Portal | ClassVar[Services]:
+) -> (
+    tractor._portal.Portal
+    |ClassVar[Services]
+):
     '''
     If no ``pikerd`` daemon-root-actor can be found start it and
     yield up (we should probably figure out returning a portal to self
@@ -281,10 +290,11 @@ async def maybe_open_pikerd(
 
     registry_addrs: list[tuple[str, int]] = (
         registry_addrs
-        or [_default_reg_addr]
+        or
+        [_default_reg_addr]
     )
 
-    pikerd_portal: tractor.Portal | None
+    pikerd_portal: tractor.Portal|None
     async with (
         open_piker_runtime(
             name=query_name,

piker/service/_daemon.py

@@ -28,6 +28,7 @@ from contextlib import (
 )
 
 import tractor
+from trio.lowlevel import current_task
 
 from ._util import (
     log,  # sub-sys logger
@@ -70,6 +71,7 @@ async def maybe_spawn_daemon(
     lock = Services.locks[service_name]
     await lock.acquire()
 
+    try:
         async with find_service(
             service_name,
             registry_addrs=[('127.0.0.1', 6116)],
@@ -134,6 +136,20 @@ async def maybe_spawn_daemon(
                 yield portal
                 await portal.cancel_actor()
 
+    except BaseException as _err:
+        err = _err
+        if (
+            lock.locked()
+            and
+            lock.statistics().owner is current_task()
+        ):
+            log.exception(
+                f'Releasing stale lock after crash..?'
+                f'{err!r}\n'
+            )
+            lock.release()
+            raise err
+
 
 async def spawn_emsd(
 

piker/service/_mngr.py

@@ -109,7 +109,7 @@ class Services:
                         # wait on any context's return value
                         # and any final portal result from the
                         # sub-actor.
-                        ctx_res: Any = await ctx.result()
+                        ctx_res: Any = await ctx.wait_for_result()
 
                         # NOTE: blocks indefinitely until cancelled
                         # either by error from the target context

piker/service/_registry.py

@@ -101,13 +101,15 @@ async def open_registry(
 
     if (
         not tractor.is_root_process()
-        and not Registry.addrs
+        and
+        not Registry.addrs
     ):
         Registry.addrs.extend(actor.reg_addrs)
 
     if (
         ensure_exists
-        and not Registry.addrs
+        and
+        not Registry.addrs
     ):
         raise RuntimeError(
             f"`{uid}` registry should already exist but doesn't?"
@@ -146,7 +148,7 @@ async def find_service(
     | list[Portal]
     | None
 ):
-
+    # try:
     reg_addrs: list[tuple[str, int]]
     async with open_registry(
         addrs=(
@@ -157,22 +159,39 @@ async def find_service(
             or Registry.addrs
         ),
     ) as reg_addrs:
-        log.info(f'Scanning for service `{service_name}`')
 
-        maybe_portals: list[Portal] | Portal | None
+        log.info(
+            f'Scanning for service {service_name!r}'
+        )
 
         # attach to existing daemon by name if possible
+        maybe_portals: list[Portal]|Portal|None
         async with tractor.find_actor(
             service_name,
             registry_addrs=reg_addrs,
             only_first=first_only,  # if set only returns single ref
         ) as maybe_portals:
             if not maybe_portals:
+                # log.info(
+                print(
+                    f'Could NOT find service {service_name!r} -> {maybe_portals!r}'
+                )
                 yield None
                 return
 
+            # log.info(
+            print(
+                f'Found service {service_name!r} -> {maybe_portals}'
+            )
             yield maybe_portals
 
+    # except BaseException as _berr:
+    #     berr = _berr
+    #     log.exception(
+    #         'tractor.find_actor() failed with,\n'
+    #     )
+    #     raise berr
+
 
 async def check_for_service(
     service_name: str,

piker/storage/__init__.py

@@ -43,7 +43,6 @@ from typing import (
 
 import numpy as np
 
-
 from .. import config
 from ..service import (
     check_for_service,
@@ -152,7 +151,10 @@ class StorageConnectionError(ConnectionError):
 
     '''
 
-def get_storagemod(name: str) -> ModuleType:
+def get_storagemod(
+    name: str,
+
+) -> ModuleType:
     mod: ModuleType = import_module(
         '.' + name,
         'piker.storage',
@@ -165,9 +167,12 @@ def get_storagemod(name: str) -> ModuleType:
 
 @acm
 async def open_storage_client(
-    backend: str | None = None,
+    backend: str|None = None,
 
-) -> tuple[ModuleType, StorageClient]:
+) -> tuple[
+    ModuleType,
+    StorageClient,
+]:
     '''
     Load the ``StorageClient`` for named backend.
 
@@ -267,7 +272,10 @@ async def open_tsdb_client(
     from ..data.feed import maybe_open_feed
 
     async with (
-        open_storage_client() as (_, storage),
+        open_storage_client() as (
+            _,
+            storage,
+        ),
 
         maybe_open_feed(
             [fqme],
@@ -275,7 +283,7 @@ async def open_tsdb_client(
 
         ) as feed,
     ):
-        profiler(f'opened feed for {fqme}')
+        profiler(f'opened feed for {fqme!r}')
 
         # to_append = feed.hist_shm.array
         # to_prepend = None

piker/ui/_app.py

@@ -21,6 +21,7 @@ Main app startup and run.
 from functools import partial
 from types import ModuleType
 
+import tractor
 import trio
 
 from piker.ui.qt import (
@@ -116,6 +117,7 @@ async def _async_main(
         needed_brokermods[brokername] = brokers[brokername]
 
     async with (
+        tractor.trionics.collapse_eg(),
         trio.open_nursery() as root_n,
     ):
         # set root nursery and task stack for spawning other charts/feeds

piker/ui/_chart.py

@@ -33,7 +33,6 @@ import trio
 
 from piker.ui.qt import (
     QtCore,
-    QtWidgets,
     Qt,
     QLineF,
     QFrame,

piker/ui/_display.py

@@ -1445,7 +1445,10 @@ async def display_symbol_data(
         # for pause/resume on mouse interaction
         rt_chart.feed = feed
 
-        async with trio.open_nursery() as ln:
+        async with (
+            tractor.trionics.collapse_eg(),
+            trio.open_nursery() as ln,
+        ):
             # if available load volume related built-in display(s)
             vlm_charts: dict[
                 str,

piker/ui/_event.py

@@ -22,7 +22,10 @@ from contextlib import asynccontextmanager as acm
 from typing import Callable
 
 import trio
-from tractor.trionics import gather_contexts
+from tractor.trionics import (
+    gather_contexts,
+    collapse_eg,
+)
 
 from piker.ui.qt import (
     QtCore,
@@ -207,7 +210,10 @@ async def open_signal_handler(
         async for args in recv:
             await async_handler(*args)
 
-    async with trio.open_nursery() as tn:
+    async with (
+        collapse_eg(),
+        trio.open_nursery() as tn
+    ):
         tn.start_soon(proxy_to_handler)
         async with send:
             yield
@@ -242,6 +248,7 @@ async def open_handlers(
     widget: QWidget
     streams: list[trio.abc.ReceiveChannel]
     async with (
+        collapse_eg(),
         trio.open_nursery() as tn,
         gather_contexts([
             open_event_stream(

piker/ui/_feedstatus.py

@@ -18,10 +18,11 @@
 Feed status and controls widget(s) for embedding in a UI-pane.
 
 """
-
 from __future__ import annotations
-from textwrap import dedent
+from typing import (
-from typing import TYPE_CHECKING
+    Any,
+    TYPE_CHECKING,
+)
 
 # from PyQt5.QtCore import Qt
 
@@ -49,35 +50,55 @@ def mk_feed_label(
     a feed control protocol.
 
     '''
-    status = feed.status
+    status: dict[str, Any] = feed.status
     assert status
 
-    msg = dedent("""
+    # SO tips on ws/nls,
-        actor: **{actor_name}**\n
+    # https://stackoverflow.com/a/15721400
-        |_ @**{host}:{port}**\n
+    ws: str = ' '
-    """)
+    # nl: str = '<br>'  # dun work?
+    actor_info_repr: str = (
+        f')> **{status["actor_short_id"]}**\n'
+        '\n'  # bc md?
+    )
 
-    for key, val in status.items():
+    # fields to select *IN* for display
-        if key in ('host', 'port', 'actor_name'):
+    # (see `.data.feed.open_feed()` status
-            continue
+    #  update -> TAG_feed_status_update)
-        msg += f'\n|_ {key}: **{{{key}}}**\n'
+    for key in [
+        'ipc',
+        'hist_shm',
+        'rt_shm',
+        'throttle_hz',
+    ]:
+        # NOTE, the 2nd key is filled via `.format()` updates.
+        actor_info_repr += (
+            f'\n'  # bc md?
+            f'{ws}|_{key}: **{{{key}}}**\n'
+        )
+        # ^TODO? formatting and content..
+        # -[ ] showing which fqme is "forward" on the
+        #    chart/fsp/order-mode?
+        #  '|_ flows: **{symbols}**\n'
+        #
+        # -[x] why isn't the indent working?
+        #  => markdown, now solved..
 
     feed_label = FormatLabel(
-        fmt_str=msg,
+        fmt_str=actor_info_repr,
-        # |_ streams: **{symbols}**\n
         font=_font.font,
         font_size=_font_small.px_size,
         font_color='default_lightest',
     )
 
+    # ?TODO, remove this?
     # form.vbox.setAlignment(feed_label, Qt.AlignBottom)
     # form.vbox.setAlignment(Qt.AlignBottom)
-    _ = chart.height() - (
+    # _ = chart.height() - (
-        form.height() +
+    #     form.height() +
-        form.fill_bar.height()
+    #     form.fill_bar.height()
-        # feed_label.height()
+    #     # feed_label.height()
-    )
+    # )
 
     feed_label.format(**feed.status)
-
     return feed_label

piker/ui/_fsp.py

@@ -600,6 +600,7 @@ async def open_fsp_admin(
             kwargs=kwargs,
         ) as (cache_hit, cluster_map),
 
+        tractor.trionics.collapse_eg(),
         trio.open_nursery() as tn,
     ):
         if cache_hit:
@@ -613,6 +614,8 @@ async def open_fsp_admin(
         )
         try:
             yield admin
+
+        # ??TODO, does this *need* to be inside a finally?
         finally:
             # terminate all tasks via signals
             for key, entry in admin._registry.items():

piker/ui/_label.py

@@ -285,18 +285,20 @@ class FormatLabel(QLabel):
         font_size: int,
         font_color: str,
 
+        use_md: bool = True,
+
         parent=None,
 
     ) -> None:
 
         super().__init__(parent)
 
-        # by default set the format string verbatim and expect user to
+        # by default set the format string verbatim and expect user
-        # call ``.format()`` later (presumably they'll notice the
+        # to call ``.format()`` later (presumably they'll notice the
         # unformatted content if ``fmt_str`` isn't meant to be
         # unformatted).
         self.fmt_str = fmt_str
-        self.setText(fmt_str)
+        # self.setText(fmt_str)  # ?TODO, why here?
 
         self.setStyleSheet(
             f"""QLabel {{
@@ -306,6 +308,7 @@ class FormatLabel(QLabel):
             """
         )
         self.setFont(_font.font)
+        if use_md:
             self.setTextFormat(
                 Qt.TextFormat.MarkdownText
             )
@@ -316,7 +319,10 @@ class FormatLabel(QLabel):
             size_policy.Expanding,
         )
         self.setAlignment(
-            Qt.AlignVCenter | Qt.AlignLeft
+            Qt.AlignLeft
+            |
+            Qt.AlignBottom
+            # Qt.AlignVCenter
         )
         self.setText(self.fmt_str)
 

piker/ui/_remote_ctl.py

@@ -15,8 +15,8 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 '''
-Remote control tasks for sending annotations (and maybe more cmds)
+Remote control tasks for sending annotations (and maybe more cmds) to
-to a chart from some other actor.
+a chart from some other actor.
 
 '''
 from __future__ import annotations
@@ -32,6 +32,7 @@ from typing import (
 )
 
 import tractor
+import trio
 from tractor import trionics
 from tractor import (
     Portal,
@@ -316,6 +317,8 @@ class AnnotCtl(Struct):
             )
             yield aid
         finally:
+            # async ipc send op
+            with trio.CancelScope(shield=True):
                 await self.remove(aid)
 
     async def redraw(

piker/ui/_search.py

@@ -15,7 +15,8 @@
 # along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 """
-qompleterz: embeddable search and complete using trio, Qt and rapidfuzz.
+qompleterz: embeddable search and complete using trio, Qt and
+rapidfuzz.
 
 """
 
@@ -46,6 +47,7 @@ import time
 from pprint import pformat
 
 from rapidfuzz import process as fuzzy
+import tractor
 import trio
 from trio_typing import TaskStatus
 
@@ -53,7 +55,7 @@ from piker.ui.qt import (
     size_policy,
     align_flag,
     Qt,
-    QtCore,
+    # QtCore,
     QtWidgets,
     QModelIndex,
     QItemSelectionModel,
@@ -920,7 +922,10 @@ async def fill_results(
 
             # issue multi-provider fan-out search request and place
             # "searching.." statuses on outstanding results providers
-            async with trio.open_nursery() as n:
+            async with (
+                tractor.trionics.collapse_eg(),
+                trio.open_nursery() as tn
+            ):
 
                 for provider, (search, pause) in (
                     _searcher_cache.copy().items()
@@ -944,7 +949,7 @@ async def fill_results(
                                 status_field='-> searchin..',
                             )
 
-                        await n.start(
+                        await tn.start(
                             pack_matches,
                             view,
                             has_results,
@@ -1004,12 +1009,14 @@ async def handle_keyboard_input(
     view.set_font_size(searchbar.dpi_font.px_size)
     send, recv = trio.open_memory_channel(616)
 
-    async with trio.open_nursery() as n:
+    async with (
-
+        tractor.trionics.collapse_eg(),  # needed?
+        trio.open_nursery() as tn
+    ):
         # start a background multi-searcher task which receives
         # patterns relayed from this keyboard input handler and
         # async updates the completer view's results.
-        n.start_soon(
+        tn.start_soon(
             partial(
                 fill_results,
                 searchw,

piker/ui/order_mode.py

@@ -555,14 +555,13 @@ class OrderMode:
 
     def on_fill(
         self,
-
         uuid: str,
         price: float,
         time_s: float,
 
         pointing: str | None = None,
 
-    ) -> None:
+    ) -> bool:
         '''
         Fill msg handler.
 
@@ -575,13 +574,33 @@ class OrderMode:
         - update fill bar size
 
         '''
-        dialog = self.dialogs[uuid]
+        # XXX WARNING XXX
+        # if a `Status(resp='error')` arrives *before* this
+        # fill-status, the `.dialogs` entry may have already been
+        # popped and thus the below will skipped.
+        #
+        # NOTE, to avoid this confusing scenario ensure that any
+        # errors delivered thru from the broker-backend are not just
+        # "noisy reporting" (like is very common from IB..) and are
+        # instead ONLY errors-causing-order-dialog-cancellation!
+        if not (dialog := self.dialogs.get(uuid)):
+            log.warning(
+                f'Order was already cleared from `.dialogs` ??\n'
+                f'uuid: {uuid!r}\n'
+            )
+            return False
+
         lines = dialog.lines
         chart = self.chart
 
-        # XXX: seems to fail on certain types of races?
+        if not lines:
+            log.warn("No line(s) for order {uuid}!?")
+            return False
+
+        # update line state(s)
+        #
+        # ?XXX this fails on certain types of races?
         # assert len(lines) == 2
-        if lines:
         flume: Flume = self.feed.flumes[chart.linked.mkt.fqme]
         _, _, ratio = flume.get_ds_info()
 
@@ -607,28 +626,31 @@ class OrderMode:
                 pointing=pointing,
                 color=lines[0].color
             )
-        else:
-            log.warn("No line(s) for order {uuid}!?")
 
     def on_cancel(
         self,
-        uuid: str
+        uuid: str,
 
-    ) -> None:
+    ) -> bool:
 
-        msg: Order = self.client._sent_orders.pop(uuid, None)
+        msg: Order|None = self.client._sent_orders.pop(uuid, None)
-
+        if msg is None:
-        if msg is not None:
-            self.lines.remove_line(uuid=uuid)
-            self.chart.linked.cursor.show_xhair()
-
-            dialog = self.dialogs.pop(uuid, None)
-            if dialog:
-                dialog.last_status_close()
-        else:
             log.warning(
                 f'Received cancel for unsubmitted order {pformat(msg)}'
             )
+            return False
+
+        # remove GUI line, show cursor.
+        self.lines.remove_line(uuid=uuid)
+        self.chart.linked.cursor.show_xhair()
+
+        # remove msg dialog (history)
+        dialog: Dialog|None = self.dialogs.pop(uuid, None)
+        if dialog:
+            dialog.last_status_close()
+
+        return True
+
 
     def cancel_orders_under_cursor(self) -> list[str]:
         return self.cancel_orders(
@@ -792,6 +814,7 @@ async def open_order_mode(
             brokerd_accounts,
             ems_dialog_msgs,
         ),
+        tractor.trionics.collapse_eg(),
         trio.open_nursery() as tn,
 
     ):
@@ -1056,13 +1079,23 @@ async def process_trade_msg(
     if name in (
         'position',
     ):
-        sym: MktPair = mode.chart.linked.mkt
+        mkt: MktPair = mode.chart.linked.mkt
         pp_msg_symbol = msg['symbol'].lower()
-        fqme = sym.fqme
+        pp_msg_bsmktid = msg['bs_mktid']
-        broker = sym.broker
+        fqme = mkt.fqme
+        broker = mkt.broker
         if (
+            # match on any backed-specific(-unique)-ID first!
+            (
+                pp_msg_bsmktid
+                and
+                mkt.bs_mktid == pp_msg_bsmktid
+            )
+            or
+            # OW try against what's provided as an FQME..
             pp_msg_symbol == fqme
-            or pp_msg_symbol == fqme.removesuffix(f'.{broker}')
+            or
+            pp_msg_symbol == fqme.removesuffix(f'.{broker}')
         ):
             log.info(
                 f'Loading position for `{fqme}`:\n'
@@ -1085,7 +1118,7 @@ async def process_trade_msg(
         return
 
     msg = Status(**msg)
-    resp = msg.resp
+    # resp: str = msg.resp
     oid = msg.oid
     dialog: Dialog = mode.dialogs.get(oid)
 
@@ -1149,19 +1182,32 @@ async def process_trade_msg(
                         mode.on_submit(oid)
 
         case Status(resp='error'):
-
-            # do all the things for a cancel:
-            # - drop order-msg dialog from client table
-            # - delete level line from view
-            mode.on_cancel(oid)
-
             # TODO: parse into broker-side msg, or should we
             # expect it to just be **that** msg verbatim (since
             # we'd presumably have only 1 `Error` msg-struct)
             broker_msg: dict = msg.brokerd_msg
+
+            # XXX NOTE, this presumes the rxed "error" is
+            # order-dialog-cancel-causing, THUS backends much ONLY
+            # relay errors of this "severity"!!
             log.error(
-                f'Order {oid}->{resp} with:\n{pformat(broker_msg)}'
+                f'Order errored ??\n'
+                f'oid: {oid!r}\n'
+                f'\n'
+                f'{pformat(broker_msg)}\n'
+                f'\n'
+                f'=> CANCELLING ORDER DIALOG <=\n'
+
+                # from tractor.devx.pformat import ppfmt
+                # !TODO LOL, wtf the msg is causing
+                # a recursion bug!
+                # -[ ] get this shit on msgspec stat!
+                # f'{ppfmt(broker_msg)}'
             )
+            # do all the things for a cancel:
+            # - drop order-msg dialog from client table
+            # - delete level line from view
+            mode.on_cancel(oid)
 
         case Status(resp='canceled'):
             # delete level line from view
@@ -1177,10 +1223,10 @@ async def process_trade_msg(
             # TODO: UX for a "pending" clear/live order
             log.info(f'Dark order triggered for {fmtmsg}')
 
-        case Status(
-            resp='triggered',
         # TODO: do the struct-msg version, blah blah..
         # req=Order(exec_mode='live', action='alert') as req,
+        case Status(
+            resp='triggered',
             req={
                 'exec_mode': 'live',
                 'action': 'alert',

tests/conftest.py

@@ -15,6 +15,12 @@ from piker.service import (
 from piker.log import get_console_log
 
 
+# include `tractor`'s built-in fixtures!
+pytest_plugins: tuple[str] = (
+    "tractor._testing.pytest",
+)
+
+
 def pytest_addoption(parser):
     parser.addoption("--ll", action="store", dest='loglevel',
                      default=None, help="logging level to set when testing")

tests/test_accounting.py

@@ -12,12 +12,14 @@ from piker import config
 from piker.accounting import (
     Account,
     calc,
-    Position,
+    open_account,
-    TransactionLedger,
-    open_trade_ledger,
     load_account,
     load_account_from_ledger,
+    open_trade_ledger,
+    Position,
+    TransactionLedger,
 )
+import tractor
 
 
 def test_root_conf_networking_section(
@@ -53,12 +55,17 @@ def test_account_file_default_empty(
 )
 def test_paper_ledger_position_calcs(
     fq_acnt: tuple[str, str],
+    debug_mode: bool,
 ):
     broker: str
     acnt_name: str
     broker, acnt_name = fq_acnt
 
-    accounts_path: Path = config.repodir() / 'tests' / '_inputs'
+    accounts_path: Path = (
+        config.repodir()
+        / 'tests'
+        / '_inputs'  # tests-local-subdir
+    )
 
     ldr: TransactionLedger
     with (
@@ -77,6 +84,7 @@ def test_paper_ledger_position_calcs(
             ledger=ldr,
 
             _fp=accounts_path,
+            debug_mode=debug_mode,
 
         ) as (dfs, ledger),
 
@@ -102,3 +110,87 @@ def test_paper_ledger_position_calcs(
         df = dfs[xrp]
         assert df['cumsize'][-1] == 0
         assert pos.cumsize == 0
+
+
+
+@pytest.mark.parametrize(
+    'fq_acnt',
+    [
+        ('ib', 'algopaper'),
+    ],
+)
+def test_ib_account_with_duplicated_mktids(
+    fq_acnt: tuple[str, str],
+    debug_mode: bool,
+):
+    # ?TODO, once we start symcache-incremental-update-support?
+    # from piker.data import (
+    #     open_symcache,
+    # )
+    #
+    # async def main():
+    #     async with (
+    #         # TODO: do this as part of `open_account()`!?
+    #         open_symcache(
+    #             'ib',
+    #             only_from_memcache=True,
+    #         ) as symcache,
+    #     ):
+
+
+    from piker.brokers.ib.ledger import (
+        tx_sort,
+
+        # ?TODO, once we want to pull lowlevel txns and process them?
+        # norm_trade_records,
+        # update_ledger_from_api_trades,
+    )
+
+    broker: str
+    acnt_id: str = 'algopaper'
+    broker, acnt_id = fq_acnt
+    accounts_def = config.load_accounts([broker])
+    assert accounts_def[f'{broker}.{acnt_id}']
+
+    ledger: TransactionLedger
+    acnt: Account
+    with (
+        tractor.devx.maybe_open_crash_handler(pdb=debug_mode),
+
+        open_trade_ledger(
+            'ib',
+            acnt_id,
+            tx_sort=tx_sort,
+
+            # TODO, eventually incrementally updated for IB..
+            # symcache=symcache,
+            symcache=None,
+            allow_from_sync_code=True,
+
+        ) as ledger,
+
+        open_account(
+            'ib',
+            acnt_id,
+            write_on_exit=True,
+        ) as acnt,
+    ):
+        # per input params
+        symcache = ledger.symcache
+        assert not (
+            symcache.pairs
+            or
+            symcache.pairs
+            or
+            symcache.mktmaps
+        )
+        # re-compute all positions that have changed state.
+        # TODO: likely we should change the API to return the
+        # position updates from `.update_from_ledger()`?
+        active, closed = acnt.dump_active()
+
+        # breakpoint()
+
+        # TODO, (see above imports as well) incremental update from
+        #      (updated) ledger?
+        # -[ ] pull some code from `.ib.broker` content.

tests/test_ems.py

@@ -42,7 +42,7 @@ from piker.accounting import (
     unpack_fqme,
 )
 from piker.accounting import (
-    open_pps,
+    open_account,
     Position,
 )
 
@@ -136,7 +136,7 @@ def load_and_check_pos(
 
 ) -> None:
 
-    with open_pps(ppmsg.broker, ppmsg.account) as table:
+    with open_account(ppmsg.broker, ppmsg.account) as table:
 
         if ppmsg.size == 0:
             assert ppmsg.symbol not in table.pps

tests/test_questrade.py

@@ -142,7 +142,12 @@ async def test_concurrent_tokens_refresh(us_symbols, loglevel):
         # async with tractor.open_nursery() as n:
         #     await n.run_in_actor('other', intermittently_refresh_tokens)
 
-        async with trio.open_nursery() as n:
+        async with (
+            tractor.trionics.collapse_eg(),
+            trio.open_nursery(
+            # strict_exception_groups=False,
+            ) as n
+        ):
 
             quoter = await qt.stock_quoter(client, us_symbols)
 
@@ -383,7 +388,9 @@ async def test_quote_streaming(tmx_symbols, loglevel, stream_what):
             else:
                 symbols = [tmx_symbols]
 
-            async with trio.open_nursery() as n:
+            async with trio.open_nursery(
+                strict_exception_groups=False,
+            ) as n:
                 for syms, func in zip(symbols, stream_what):
                     n.start_soon(func, feed, syms)