coorection on switches

CHANGELOG.md

@@ -1,5 +1,30 @@
 # Changelog
 
+## 0.3.1
+
+### Fixed
+
+- **A paused registration is no longer silent.** It stops claiming and nothing
+  said why, so the only symptom was work that did not happen — and the switch
+  registry is shared, so a switch left set by another environment, an old test
+  run or an operator months ago stops work in a process nobody has touched.
+  Diagnosing it meant reading a Redis hash by hand.
+
+  Boot now names anything in *this* process's topology that starts paused, and a
+  flip is logged when it happens or when it is next noticed:
+
+  ```
+  WARNING 1 registration(s) start PAUSED by a switch: catalog
+  WARNING switch: paused catalog
+  INFO    switch: resumed catalog
+  ```
+
+  Switches set for services this process does not run are ignored, so the
+  warning stays worth reading.
+
+- The topology log printed `<Topic.X: 'x'>` for names that are `StrEnum` members
+  — which is how a typical app names them — instead of the plain name.
+
 ## 0.3.0
 
 A clean break. The public API is replaced, and the Redis keyspace changes, so

runtime/_version.py

@@ -1 +1 @@
-__version__ = "0.3.0"
+__version__ = "0.3.1"

runtime/app.py

@@ -213,6 +213,35 @@ def _slot_count(self) -> int:
             for inc in self._services
         )
 
+    def _warn_if_paused(self, switches: SwitchBoard) -> None:
+        """Name anything this process is about to start that is already paused.
+
+        The switch registry is shared, so a switch left set by another
+        environment, a test run or an operator months ago silently stops work
+        here with no other symptom. Boot is the one moment we can say so against
+        the actual topology, rather than listing names that may belong to someone
+        else entirely.
+        """
+        paused = sorted(
+            {
+                name
+                for inc in self._services
+                for name in (
+                    inc.service.name,
+                    *(reg.name for reg in inc.service.consumers),
+                    *(reg.id for reg in inc.service._every),
+                    *(reg.id for reg in inc.service._once),
+                )
+                if not switches.is_enabled(name)
+            }
+        )
+        if paused:
+            log.warning(
+                "%d registration(s) start PAUSED by a switch: %s",
+                len(paused),
+                ", ".join(paused),
+            )
+
     def _log_topology(self) -> None:
         _ensure_logging()
         log.info("App(redis=%s, namespace=%r) — topology:", self.redis, self.namespace)
@@ -239,7 +268,9 @@ def _log_topology(self) -> None:
                 log.info(
                     "    %s %r → stream %r%s slots=%d (%s)",
                     kind,
-                    reg.name,
+                    # str(): names are routinely StrEnum members, whose repr is
+                    # <Topic.X: 'x'> and makes the topology hard to read.
+                    str(reg.name),
                     reg.stream,
                     group,
                     slots,
@@ -282,6 +313,7 @@ async def _serve(self, broker: Broker | None = None) -> None:
         node = Node(bk)
         switches = SwitchBoard(bk, interval_s=self.switch_interval)
         await switches.refresh()  # start with the truth, not with "all enabled"
+        self._warn_if_paused(switches)
 
         try:
             async with AsyncExitStack() as stack:

runtime/switches.py

@@ -72,9 +72,21 @@ async def refresh(self) -> None:
             # Redis and is strictly newer than what we are holding, so publishing
             # this would silently undo it.
             return
-        self._paused = frozenset(
-            name for name, enabled in switches.items() if not enabled
-        )
+        paused = frozenset(name for name, enabled in switches.items() if not enabled)
+        self._log_changes(paused)
+        self._paused = paused
+
+    def _log_changes(self, paused: frozenset[str]) -> None:
+        """Say so when a switch flips.
+
+        A paused consumer is otherwise completely silent — it stops claiming and
+        nothing anywhere says why — so the only symptom is work that does not
+        happen. That is a miserable thing to diagnose from the outside.
+        """
+        if newly := paused - self._paused:
+            log.warning("switch: paused %s", ", ".join(sorted(newly)))
+        if resumed := self._paused - paused:
+            log.info("switch: resumed %s", ", ".join(sorted(resumed)))
 
     async def run(self) -> None:
         """Refresh forever. Runs as an App task."""
@@ -86,7 +98,12 @@ async def set(self, name: str, enabled: bool) -> None:
         """Flip a switch and apply it locally at once, so the caller's own next
         check reflects what it just did instead of waiting for the timer."""
         await self._broker.switch_set(name, enabled)
-        self._paused = self._paused - {name} if enabled else self._paused | {name}
+        paused = self._paused - {name} if enabled else self._paused | {name}
+        # Logged here as well as in refresh(): a flip made from inside this
+        # process never goes through a re-read, so it would otherwise be the one
+        # kind of pause that leaves no trace at all.
+        self._log_changes(paused)
+        self._paused = paused
         self._generation += 1
 
 

tests/test_app.py

@@ -232,6 +232,89 @@ async def doctor(ctx):
     assert handled == [], "the action should be paused by the master switch"
 
 
+async def test_boot_names_anything_that_starts_paused(broker, caplog):
+    """A paused consumer is otherwise silent: it stops claiming and nothing says
+    why. A switch left set by another environment or an old test run then stops
+    work here with no symptom but absence."""
+    svc = Service("catalog", max_slots=1)
+    svc.register_action(noop, name="catalog.sync")
+    svc.register_every(lambda ctx: asyncio.sleep(0), interval="1s", id="doctor")
+    await broker.switch_set("catalog", False)
+
+    app = App(redis="unused://", switch_interval=0.02)
+    app.include(svc)
+
+    with caplog.at_level("WARNING", logger="runtime"):
+        async with serving(app, broker):
+            await asyncio.sleep(0.1)
+
+    assert "start PAUSED" in caplog.text
+    assert "catalog" in caplog.text
+
+
+async def test_boot_says_nothing_when_nothing_is_paused(broker, caplog):
+    svc = Service("catalog", max_slots=1)
+    svc.register_action(noop, name="catalog.sync")
+
+    app = App(redis="unused://", switch_interval=0.02)
+    app.include(svc)
+
+    with caplog.at_level("WARNING", logger="runtime"):
+        async with serving(app, broker):
+            await asyncio.sleep(0.1)
+
+    assert "start PAUSED" not in caplog.text
+
+
+async def test_boot_ignores_switches_for_things_this_process_does_not_run(
+    broker, caplog
+):
+    """The registry is shared. Listing someone else's paused service as if it
+    affected this process would make the warning noise, and noise gets ignored."""
+    svc = Service("catalog", max_slots=1)
+    svc.register_action(noop, name="catalog.sync")
+    await broker.switch_set("some.other.deployment", False)
+
+    app = App(redis="unused://", switch_interval=0.02)
+    app.include(svc)
+
+    with caplog.at_level("WARNING", logger="runtime"):
+        async with serving(app, broker):
+            await asyncio.sleep(0.1)
+
+    assert "start PAUSED" not in caplog.text
+
+
+async def test_a_local_switch_flip_is_logged(broker, caplog):
+    board = SwitchBoard(broker)
+
+    with caplog.at_level("INFO", logger="runtime"):
+        await board.set("catalog", False)
+        assert "paused catalog" in caplog.text
+
+        await board.set("catalog", True)
+
+    assert "resumed catalog" in caplog.text
+
+
+async def test_a_flip_made_elsewhere_is_logged_when_it_is_noticed(broker, caplog):
+    """The important direction: an operator's redis-cli, another process, or a
+    switch left set by a previous deployment. Nothing local knows it happened
+    until the next refresh."""
+    board = SwitchBoard(broker)
+    await board.refresh()
+
+    with caplog.at_level("INFO", logger="runtime"):
+        await broker.switch_set("catalog", False)  # bypassing this board entirely
+        await board.refresh()
+        assert "paused catalog" in caplog.text
+
+        await broker.switch_set("catalog", True)
+        await board.refresh()
+
+    assert "resumed catalog" in caplog.text
+
+
 async def test_a_failed_refresh_keeps_the_previous_snapshot(broker):
     """Losing Redis must not silently resume everything that was paused."""