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

tests/test_log_sys.py

@@ -4,6 +4,7 @@
 '''
 from pathlib import Path
 import shutil
+from types import ModuleType
 
 import pytest
 import tractor
@@ -57,28 +58,32 @@ def test_implicit_mod_name_applied_for_child(
     mod_code: str = (
         f'import tractor\n'
         f'\n'
-        # f'breakpoint()\n'  # if you want to trace it all
+        # if you need to trace `testdir` stuff @ import-time..
+        # f'breakpoint()\n'
         f'log = tractor.log.get_logger(pkg_name="{proj_name}")\n'
     )
 
     # create a sub-module for each pkg layer
     _lib = testdir.mkpydir(proj_name)
     pkg: Path = Path(_lib)
+    pkg_init_mod: Path = pkg / "__init__.py"
+    pkg_init_mod.write_text(mod_code)
+
     subpkg: Path = pkg / 'subpkg'
     subpkg.mkdir()
-
-    pkgmod: Path = subpkg / "__init__.py"
-    pkgmod.touch()
+    subpkgmod: Path = subpkg / "__init__.py"
+    subpkgmod.touch()
+    subpkgmod.write_text(mod_code)
 
     _submod: Path = testdir.makepyfile(
         _mod=mod_code,
     )
 
-    pkg_mod = pkg / 'mod.py'
+    pkg_submod = pkg / 'mod.py'
     pkg_subpkg_submod = subpkg / 'submod.py'
     shutil.copyfile(
         _submod,
-        pkg_mod,
+        pkg_submod,
     )
     shutil.copyfile(
         _submod,
@@ -91,10 +96,11 @@ def test_implicit_mod_name_applied_for_child(
     # XXX NOTE, once the "top level" pkg mod has been
     # imported, we can then use `import` syntax to
     # import it's sub-pkgs and modules.
-    pkgmod = _code_load.load_module_from_path(
+    subpkgmod: ModuleType = _code_load.load_module_from_path(
         Path(pkg / '__init__.py'),
         module_name=proj_name,
     )
+
     pkg_root_log = log.get_logger(
         pkg_name=proj_name,
         mk_sublog=False,
@@ -107,16 +113,31 @@ def test_implicit_mod_name_applied_for_child(
     # ^TODO! test this same output but created via a `get_logger()`
     # call in the `snakelib.__init__py`!!
 
-    # a first-pkg-level module should only
-    # use
+    # NOTE, the pkg-level "init mod" should of course
+    # have the same name as the package ns-path.
+    import snakelib as init_mod
+    assert init_mod.log.name == proj_name
+
+    # NOTE, a first-pkg-level sub-module should only
+    # use the package-name since the leaf-node-module
+    # will be included in log headers by default.
     from snakelib import mod
     assert mod.log.name == proj_name
 
+    from snakelib import subpkg
+    assert (
+        subpkg.log.name
+        ==
+        subpkg.__package__ 
+        ==
+        f'{proj_name}.subpkg'
+    )
+
     from snakelib.subpkg import submod
     assert (
         submod.log.name
         ==
-        submod.__package__  # ?TODO, use this in `.get_logger()` instead?
+        submod.__package__ 
         ==
         f'{proj_name}.subpkg'
     )
@@ -125,8 +146,6 @@ def test_implicit_mod_name_applied_for_child(
     assert len(sub_logs) == 1  # only one nested sub-pkg module
     assert submod.log.logger in sub_logs
 
-    # breakpoint()
-
 
 # TODO, moar tests against existing feats:
 # ------ - ------

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