Unmask `ClosedResourceError` handling in `._transport`

Unmask the CRE case block for peer-closed socket errors which already
had a TODO about reproducing the condition. It appears this case can
happen during inter-actor comms teardowns in `piker`, but i haven't been
able to figure out exactly what reproduces it yet..

So activate the block again for that 'socket already closed'-msg case,
and add a TODO questioning how to reproduce it.

(this commit msg was generated in some part by [`claude-code`][claude-code-gh])
[claude-code-gh]: https://github.com/anthropics/claude-code

tractor/_rpc.py

@@ -284,6 +284,10 @@ async def _errors_relayed_via_ipc(
     try:
         yield  # run RPC invoke body
 
+    except TransportClosed:
+        log.exception('Tpt disconnect during remote-exc relay?')
+        raise
+
     # box and ship RPC errors for wire-transit via
     # the task's requesting parent IPC-channel.
     except (
@@ -319,6 +323,9 @@ async def _errors_relayed_via_ipc(
                         and debug_kbis
                     )
                 )
+                # TODO? better then `debug_filter` below?
+                and
+                not isinstance(err, TransportClosed)
             ):
                 # XXX QUESTION XXX: is there any case where we'll
                 # want to debug IPC disconnects as a default?
@@ -327,13 +334,25 @@ async def _errors_relayed_via_ipc(
                 # recovery logic - the only case is some kind of
                 # strange bug in our transport layer itself? Going
                 # to keep this open ended for now.
-                log.debug(
+
-                    'RPC task crashed, attempting to enter debugger\n'
+                if _state.debug_mode():
-                    f'|_{ctx}'
+                    log.exception(
-                )
+                        f'RPC task crashed!\n'
+                        f'Attempting to enter debugger\n'
+                        f'\n'
+                        f'{ctx}'
+                    )
+
                 entered_debug = await debug._maybe_enter_pm(
                     err,
                     api_frame=inspect.currentframe(),
+
+                    # don't REPL any psuedo-expected tpt-disconnect
+                    # debug_filter=lambda exc: (
+                    #     type (exc) not in {
+                    #         TransportClosed,
+                    #     }
+                    # ),
                 )
                 if not entered_debug:
                     # if we prolly should have entered the REPL but
@@ -450,7 +469,7 @@ async def _invoke(
     kwargs: dict[str, Any],
 
     is_rpc: bool = True,
-    hide_tb: bool = True,
+    hide_tb: bool = False,
     return_msg_type: Return|CancelAck = Return,
 
     task_status: TaskStatus[
@@ -675,6 +694,22 @@ async def _invoke(
                     f'{pretty_struct.pformat(return_msg)}\n'
                 )
                 await chan.send(return_msg)
+                # ?TODO, remove the below since .send() already
+                # doesn't raise on tpt-closed?
+                # try:
+                #     await chan.send(return_msg)
+                # except TransportClosed:
+                #     log.exception(
+                #         f"Failed send final result to 'parent'-side of IPC-ctx!\n"
+                #         f'\n'
+                #         f'{chan}\n'
+                #         f'Channel already disconnected ??\n'
+                #         f'\n'
+                #         f'{pretty_struct.pformat(return_msg)}'
+                #     )
+                #     # ?TODO? will this ever be true though?
+                #     if chan.connected():
+                #         raise
 
             # NOTE: this happens IFF `ctx._scope.cancel()` is
             # called by any of,

tractor/ipc/_transport.py

@@ -459,21 +459,23 @@ class MsgpackTransport(MsgTransport):
                         )
                         raise tpt_closed from trans_err
 
-                    # case trio.ClosedResourceError() if (
+                    # ??TODO??, what case in piker does this and HOW
-                    #     'this socket was already closed'
+                    # CAN WE RE-PRODUCE IT?!?!?
-                    #     in
+                    case trio.ClosedResourceError() if (
-                    #     trans_err.args[0]
+                        'this socket was already closed'
-                    # ):
+                        in
-                    #     tpt_closed = TransportClosed.from_src_exc(
+                        trans_err.args[0]
-                    #         message=(
+                    ):
-                    #             f'{tpt_name} already closed by peer\n'
+                        tpt_closed = TransportClosed.from_src_exc(
-                    #         ),
+                            message=(
-                    #         body=f'{self}\n',
+                                f'{tpt_name} already closed by peer\n'
-                    #         src_exc=trans_err,
+                            ),
-                    #         raise_on_report=True,
+                            body=f'{self}\n',
-                    #         loglevel='transport',
+                            src_exc=trans_err,
-                    #     )
+                            raise_on_report=True,
-                    #     raise tpt_closed from trans_err
+                            loglevel='transport',
+                        )
+                        raise tpt_closed from trans_err
 
                     # unless the disconnect condition falls under "a
                     # normal operation breakage" we usualy console warn