diff --git a/CHANGELOG.md b/CHANGELOG.md index 977d1f7..210dd85 100644 --- a/CHANGELOG.md +++ b/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 `` 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 diff --git a/runtime/_version.py b/runtime/_version.py index 493f741..260c070 100644 --- a/runtime/_version.py +++ b/runtime/_version.py @@ -1 +1 @@ -__version__ = "0.3.0" +__version__ = "0.3.1" diff --git a/runtime/app.py b/runtime/app.py index 30f5d58..2ca3c26 100644 --- a/runtime/app.py +++ b/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 + # 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: diff --git a/runtime/switches.py b/runtime/switches.py index 1201427..34e2197 100644 --- a/runtime/switches.py +++ b/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 diff --git a/tests/test_app.py b/tests/test_app.py index eefd8bf..d66c147 100644 --- a/tests/test_app.py +++ b/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."""