refactor: one typed endmarker sentinel, collapsing make_receive_queue

make_receive_queue branched on its endmarker only to decide between
calling setcallback with and without it -- which looks redundant, because
setcallback's default *is* NO_ENDMARKER_WANTED.  It was not: _multi and
_channel each defined their own `object()` under that name, and the
consumer task compares against _channel's.  Passing _multi's through
would have been read as a real endmarker and queued a bare object.

So share one sentinel and name its type.  It is an enum member rather
than a bare object() so it can appear in an annotation: an endmarker may
be any object, identity is the only thing separating "none wanted" from a
legitimate one, and `Endmarker = object | Literal[NoEndmarker.NOT_WANTED]`
now says on every signature which sentinel a caller has to hand back.
With one definition the branch collapses to a plain forward.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: RonnyPfannschmidt

src/execnet/_channel.py

@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import enum
 import threading
 import weakref
 from collections.abc import Callable
@@ -30,7 +31,26 @@
 if TYPE_CHECKING:
     from ._gateway_base import BaseGateway
 
-NO_ENDMARKER_WANTED = object()
+
+class NoEndmarker(enum.Enum):
+    """Type of the "no endmarker wanted" sentinel.
+
+    An enum rather than a bare ``object()`` so it is nameable in an
+    annotation: an endmarker may be *any* object, so the only thing that
+    distinguishes "none wanted" from a legitimate endmarker is identity,
+    and a second module inventing its own ``object()`` for it would be
+    silently wrong.  ``endmarker: object | Literal[NoEndmarker.NOT_WANTED]``
+    says which sentinel a caller has to hand back.
+    """
+
+    NOT_WANTED = enum.auto()
+
+
+NO_ENDMARKER_WANTED = NoEndmarker.NOT_WANTED
+
+#: what an ``endmarker=`` parameter accepts: any object to deliver at the
+#: end, or the sentinel meaning "do not deliver one"
+Endmarker = object | Literal[NoEndmarker.NOT_WANTED]
 
 
 class Channel:
@@ -82,7 +102,7 @@ def _trace(self, *msg: object) -> None:
     def setcallback(
         self,
         callback: Callable[[Any], Any],
-        endmarker: object = NO_ENDMARKER_WANTED,
+        endmarker: Endmarker = NO_ENDMARKER_WANTED,
     ) -> None:
         """Set a callback function for receiving items.
 

src/execnet/_gateway_base.py

@@ -27,6 +27,7 @@
 from ._boundary import make_wakener
 from ._channel import Channel
 from ._channel import ChannelFactory
+from ._channel import Endmarker
 from ._errors import INTERRUPT_TEXT
 from ._errors import geterrortext
 from ._errors import sysex
@@ -109,7 +110,7 @@ def _start_channel_consumer(
         self,
         channel: Channel,
         callback: Callable[[Any], Any],
-        endmarker: object,
+        endmarker: Endmarker,
     ) -> None:
         """Attach a receiver callback: hand the channel to a consumer task.
 

src/execnet/_multi.py

@@ -25,7 +25,9 @@
 from typing import overload
 
 from ._boundary import WaitBackend
+from ._channel import NO_ENDMARKER_WANTED
 from ._channel import Channel
+from ._channel import Endmarker
 from ._execmodel import ExecModel
 from ._execmodel import get_execmodel
 from ._execmodel import resolve_profile
@@ -39,9 +41,6 @@
     from ._gateway import Gateway
 
 
-NO_ENDMARKER_WANTED = object()
-
-
 class Group:
     """Gateway Group."""
 
@@ -379,7 +378,7 @@ def receive_each(
                 l.append(obj)
         return l
 
-    def make_receive_queue(self, endmarker: object = NO_ENDMARKER_WANTED):
+    def make_receive_queue(self, endmarker: Endmarker = NO_ENDMARKER_WANTED):
         try:
             return self._queue  # type: ignore[has-type]
         except AttributeError:
@@ -391,10 +390,7 @@ def make_receive_queue(self, endmarker: object = NO_ENDMARKER_WANTED):
                 def putreceived(obj, channel: Channel = ch) -> None:
                     self._queue.put((channel, obj))  # type: ignore[union-attr]
 
-                if endmarker is NO_ENDMARKER_WANTED:
-                    ch.setcallback(putreceived)
-                else:
-                    ch.setcallback(putreceived, endmarker=endmarker)
+                ch.setcallback(putreceived, endmarker=endmarker)
             return self._queue
 
     def waitclose(self) -> None:

src/execnet/_trio_host.py

@@ -28,6 +28,7 @@
 from ._boundary import Flag
 from ._channel import ENDMARKER
 from ._channel import NO_ENDMARKER_WANTED
+from ._channel import Endmarker
 from ._errors import GatewayReceivedTerminate
 from ._errors import RemoteError
 from ._execmodel import ExecModel
@@ -315,7 +316,7 @@ def attach_consumer(
         self,
         channel: Any,
         callback: Callable[[Any], Any],
-        endmarker: object,
+        endmarker: Endmarker,
     ) -> None:
         """Switch ``channel`` to callback mode: a loop task drains it.
 
@@ -394,7 +395,7 @@ async def _run_consumer(
         channel: Any,
         inbox: trio.MemoryReceiveChannel[bytes],
         callback: Callable[[Any], Any],
-        endmarker: object,
+        endmarker: Endmarker,
         done: Flag,
     ) -> None:
         """Drain ``inbox`` into ``callback`` (each call off the loop thread).

testing/test_cli.py

@@ -14,7 +14,7 @@
 import subprocess
 import sys
 import tempfile
-from collections.abc import Iterator
+from typing import Any
 
 import pytest
 
@@ -30,7 +30,7 @@
 
 
 def worker_config(**overrides: object) -> str:
-    config = {
+    config: dict[str, object] = {
         "id": "cli-test-worker",
         "profile": "thread",
         "execmodel": "thread",
@@ -89,9 +89,7 @@ def test_config_sources_are_mutually_exclusive(self) -> None:
                 ["worker", "--config", "{}", "--config-fd", "0"]
             )
 
-    @pytest.mark.parametrize(
-        ("value", "expected"), [("3", (3,)), ("4,5", (4, 5))]
-    )
+    @pytest.mark.parametrize(("value", "expected"), [("3", (3,)), ("4,5", (4, 5))])
     def test_protocol_fd_accepts_one_fd_or_a_pair(
         self, value: str, expected: tuple[int, ...]
     ) -> None:
@@ -110,7 +108,7 @@ def test_protocol_fd_rejects_nonsense(self) -> None:
             (":8888", ("tcp", ("localhost", 8888))),
         ],
     )
-    def test_parse_address(self, address: str, expected: tuple) -> None:
+    def test_parse_address(self, address: str, expected: tuple[str, Any]) -> None:
         from execnet._trio_worker import parse_address
 
         assert parse_address(address) == expected
@@ -168,6 +166,7 @@ def test_a_plain_pipe_fd_is_rejected(self) -> None:
                 capture_output=True,
                 text=True,
                 timeout=TESTTIMEOUT,
+                check=False,
             )
         finally:
             os.close(read_fd)
@@ -208,9 +207,7 @@ def test_config_fd_keeps_it_out_of_argv(self) -> None:
     def test_config_file(self, tmp_path) -> None:
         path = tmp_path / "config.json"
         path.write_text(worker_config())
-        ns = _cli._build_parser().parse_args(
-            ["worker", "--config-file", str(path)]
-        )
+        ns = _cli._build_parser().parse_args(["worker", "--config-file", str(path)])
         assert _cli._load_config(ns)["id"] == "cli-test-worker"
 
     def test_no_config_source_is_an_error(self) -> None:
@@ -226,13 +223,15 @@ def test_defaults_per_platform(self) -> None:
         assert _provision.resolve_transport(spec) == expected
 
     def test_explicit_wins(self) -> None:
-        assert _provision.resolve_transport(execnet.XSpec("popen//transport=stdio")) == (
-            "stdio"
-        )
+        assert _provision.resolve_transport(
+            execnet.XSpec("popen//transport=stdio")
+        ) == ("stdio")
 
     def test_unknown_is_rejected(self) -> None:
         with pytest.raises(ValueError, match="unknown transport"):
-            _provision.resolve_transport(execnet.XSpec("popen//transport=carrier-pigeon"))
+            _provision.resolve_transport(
+                execnet.XSpec("popen//transport=carrier-pigeon")
+            )
 
     @posix_only
     def test_socket_transport_keeps_the_protocol_off_stdio(self) -> None:
@@ -353,7 +352,7 @@ def test_server_is_the_socketserver(self) -> None:
 
     def test_socketserver_alias_warns(self, monkeypatch: pytest.MonkeyPatch) -> None:
         called: list[list[str]] = []
-        monkeypatch.setattr(_cli, "main", lambda argv: called.append(argv))
+        monkeypatch.setattr(_cli, "main", called.append)
         with pytest.warns(DeprecationWarning, match="execnet server"):
             _cli.socketserver_main([":0", "--once"])
         assert called == [["server", ":0", "--once"]]