diff --git a/config.ini.example b/config.ini.example index 751b18d..c60f70d 100644 --- a/config.ini.example +++ b/config.ini.example @@ -1888,7 +1888,8 @@ iata = XYZ # # meshcore/{IATA}/{PUBLIC_KEY}/neighbors), else # # /neighbors. A derived # # location-routed topic is skipped with a warning when -# # no iata is set, rather than publishing to XYZ. +# # no iata is set (blank or XYZ), rather than publishing +# # into that shared namespace. # # Topic template placeholders: # {IATA} - Uppercase IATA code (e.g., SEA) diff --git a/modules/commands/neighbors_command.py b/modules/commands/neighbors_command.py index 0e3b3ec..00f93ac 100644 --- a/modules/commands/neighbors_command.py +++ b/modules/commands/neighbors_command.py @@ -12,6 +12,11 @@ from typing import Any, Optional from ..models import MeshMessage from .base_command import BaseCommand +# How long a refused sender must wait before retrying. Matches the default +# discover window: long enough that a busy cycle is usually done, short enough +# that "busy" / "disabled" is not a fifteen-minute lockout. +_REFUSAL_RETRY_SECONDS = 60.0 + class NeighborsCommand(BaseCommand): """Runs one neighbor discovery cycle immediately. @@ -107,13 +112,26 @@ class NeighborsCommand(BaseCommand): self.logger.debug(f"Neighbors: could not read the shared cooldown: {e}") return 0.0 + def _busy_retry_seconds(self, service: Any) -> float: + """How long a 'busy' refusal should hold this sender's personal cooldown. + + Aligns with the discover window when known: by then a normal cycle has + finished listening, so a retry is meaningful. + """ + cfg = getattr(service, 'neighbors_config', None) + window = getattr(cfg, 'discover_window', None) + if isinstance(window, (int, float)) and window > 0: + return float(window) + return _REFUSAL_RETRY_SECONDS + def _yield_user_cooldown(self, user_id: Optional[str], remaining: float) -> None: - """Rewind this sender's own cooldown to expire with the shared one. + """Rewind this sender's own cooldown to expire after *remaining* seconds. The command manager records the execution *before* calling execute(), so a request refused in here has already spent the sender's 15 minutes. Without - this, "wait 1 more minute" would be a lie: retrying a minute later would - be refused for another fourteen, by their personal cooldown this time. + this, "wait 1 more minute" (or "busy, try again shortly") would be a lie: + retrying when ready would be refused for another fourteen minutes by their + personal cooldown. Rewound rather than cleared, so the refusal itself cannot be spammed — a DM reply is airtime too. @@ -134,16 +152,22 @@ class NeighborsCommand(BaseCommand): """ service = self._get_capture_service() if service is None or not getattr(service, 'neighbors_enabled', False): + # Same yield as other refusals: the manager already recorded a full + # personal cooldown, and "disabled" must not become a 15-minute lockout + # after the operator turns the feature on. + self._yield_user_cooldown(message.sender_id, _REFUSAL_RETRY_SECONDS) await self.send_response(message, self.translate('commands.neighbors.disabled')) return True if self._cycle_task is not None and not self._cycle_task.done(): + self._yield_user_cooldown(message.sender_id, self._busy_retry_seconds(service)) await self.send_response(message, self.translate('commands.neighbors.busy')) return True # The service refuses an overlapping cycle on its own; this only decides # what the requester is told, and rations airtime across senders. if getattr(service, 'neighbors_cycle_active', False): + self._yield_user_cooldown(message.sender_id, self._busy_retry_seconds(service)) await self.send_response(message, self.translate('commands.neighbors.busy')) return True @@ -163,22 +187,69 @@ class NeighborsCommand(BaseCommand): cfg = service.neighbors_config self.logger.info(f"User {message.sender_id} requested a neighbors discovery cycle") - await self.send_response( - message, - self.translate('commands.neighbors.started', seconds=int(cfg.discover_window)), - ) + + # Claim before acknowledging. send_response awaits, and without the lock + # the scheduler can start a cycle in that gap — the user would then get + # "started" followed by "failed: already running". + claimed = False + claim = getattr(service, 'claim_neighbors_cycle', None) + if callable(claim): + refusal = claim() + if refusal is not None: + if "already" in refusal: + self._yield_user_cooldown( + message.sender_id, self._busy_retry_seconds(service) + ) + await self.send_response( + message, self.translate('commands.neighbors.busy') + ) + else: + left = self._shared_cooldown_remaining(service) or _REFUSAL_RETRY_SECONDS + self._yield_user_cooldown(message.sender_id, left) + await self.send_response( + message, + self.translate( + 'commands.neighbors.cooldown_active', + minutes=max(1, math.ceil(left / 60)), + ), + ) + return True + claimed = True + + try: + await self.send_response( + message, + self.translate('commands.neighbors.started', seconds=int(cfg.discover_window)), + ) + except Exception: + release = getattr(service, 'release_neighbors_cycle', None) + if claimed and callable(release): + release() + raise # Run detached so the discover window does not hold the command open, and # bound it so a stalled radio link cannot leave the task alive forever. self._cycle_task = asyncio.create_task( - self._run_and_report(message, service, cfg.cycle_budget) + self._run_and_report( + message, service, cfg.cycle_budget, already_claimed=claimed + ) ) return True - async def _run_and_report(self, message: MeshMessage, service: Any, budget: float) -> None: + async def _run_and_report( + self, + message: MeshMessage, + service: Any, + budget: float, + *, + already_claimed: bool = False, + ) -> None: """Run one cycle and DM the outcome.""" try: - summary = await asyncio.wait_for(service.run_neighbors_cycle(), timeout=budget) + summary = await asyncio.wait_for( + service.run_neighbors_cycle(already_claimed=already_claimed), + timeout=budget, + ) except asyncio.TimeoutError: self.logger.error(f"Neighbors: manual cycle exceeded {budget:.0f}s and was abandoned") await self.send_response( diff --git a/modules/service_plugins/packet_capture_service.py b/modules/service_plugins/packet_capture_service.py index f0549d2..509d535 100644 --- a/modules/service_plugins/packet_capture_service.py +++ b/modules/service_plugins/packet_capture_service.py @@ -461,9 +461,12 @@ class PacketCaptureService(BaseServicePlugin): self.mqtt_enabled = config.getboolean("PacketCapture", "mqtt_enabled", fallback=True) self.mqtt_brokers = self._parse_mqtt_brokers(config) - # Global IATA - configured_iata = config.get("PacketCapture", "iata", fallback=DEFAULT_IATA) - self.global_iata = (configured_iata.strip() or DEFAULT_IATA).lower() + # Global IATA. Blank stays blank so packet/status {IATA} topics keep + # their historical empty-segment resolution; a missing key still falls + # back to the XYZ sentinel. Neighbors treats blank and XYZ as unset. + self.global_iata = config.get( + "PacketCapture", "iata", fallback=DEFAULT_IATA + ).strip().lower() # Owner information self.owner_public_key = config.get("PacketCapture", "owner_public_key", fallback=None) @@ -2135,23 +2138,27 @@ class PacketCaptureService(BaseServicePlugin): return f"{prefix}/neighbors" return None + def _iata_is_unset(self) -> bool: + """True when no real IATA is configured (blank or the XYZ sentinel).""" + return (not self.global_iata) or self.global_iata == DEFAULT_IATA.lower() + def _resolve_neighbors_topic(self, broker_config: dict[str, Any]) -> Optional[str]: """Resolved topic for one broker's neighbors snapshot, or None if unroutable.""" template = self._neighbors_topic_template(broker_config) if not template: return None - # An unset IATA is the documented sentinel "XYZ", and this topic is - # location-routed on the community brokers. Publishing a snapshot to - # meshcore/XYZ/... would pollute a shared namespace, so refuse instead — - # matching the upstream rule that neighbors needs a real IATA to route. - # Only applies to a template we derived; an explicit topic is the - # operator's call. + # An unset IATA is blank or the documented sentinel "XYZ", and this topic + # is location-routed on the community brokers. Publishing a snapshot to + # meshcore/XYZ/... (or meshcore//...) would pollute a shared namespace, so + # refuse instead — matching the upstream rule that neighbors needs a real + # IATA to route. Only applies to a template we derived; an explicit topic + # is the operator's call. # .upper() catches both the {IATA} and {iata} placeholder spellings. if ( not broker_config.get("topic_neighbors") and "{IATA}" in template.upper() - and self.global_iata == DEFAULT_IATA.lower() + and self._iata_is_unset() ): return None @@ -2293,7 +2300,39 @@ class PacketCaptureService(BaseServicePlugin): return 0.0 return max(0.0, MIN_CYCLE_GAP_SECONDS - (time.time() - last)) - async def run_neighbors_cycle(self) -> dict[str, Any]: + def claim_neighbors_cycle(self) -> Optional[str]: + """Claim the single-flight lock before a cycle reaches the radio. + + Returns ``None`` when claimed. Returns a refusal reason when another + cycle is already running or the airtime cooldown has not expired. + + The ``neighbors`` command claims *before* acknowledging so an await + cannot let the scheduler sneak in and turn "started" into an immediate + failure. Pair every successful claim with ``release_neighbors_cycle`` + (``run_neighbors_cycle`` does this in ``finally``). + """ + if self.neighbors_cycle_active: + self.logger.info( + "Neighbors: a discovery cycle is already running, skipping this trigger" + ) + return "a discovery cycle is already running" + + cooldown = self.neighbors_cooldown_remaining() + if cooldown > 0: + self.logger.info( + f"Neighbors: last cycle was too recent, {cooldown:.0f}s left before " + f"another may run" + ) + return f"another cycle may run in {cooldown:.0f}s" + + self.neighbors_cycle_active = True + return None + + def release_neighbors_cycle(self) -> None: + """Drop the single-flight lock claimed by ``claim_neighbors_cycle``.""" + self.neighbors_cycle_active = False + + async def run_neighbors_cycle(self, *, already_claimed: bool = False) -> dict[str, Any]: """Run one discovery cycle, subject to the airtime guards. Both guards live here rather than in a caller, because the scheduler, the @@ -2304,31 +2343,23 @@ class PacketCaptureService(BaseServicePlugin): * no cycle within ``MIN_CYCLE_GAP_SECONDS`` of the last one that transmitted. + Pass ``already_claimed=True`` when the caller has successfully called + ``claim_neighbors_cycle`` (the manual command does this so it can + acknowledge only after the lock is held). + Returns a summary dict (also used by the ``neighbors`` command): ``{'ok', 'reason', 'discovered', 'queried', 'best_snr', 'attempted', 'succeeded', 'recorded'}``. """ - if self.neighbors_cycle_active: - self.logger.info( - "Neighbors: a discovery cycle is already running, skipping this trigger" - ) - return self._empty_neighbors_summary("a discovery cycle is already running") + if not already_claimed: + reason = self.claim_neighbors_cycle() + if reason is not None: + return self._empty_neighbors_summary(reason) - cooldown = self.neighbors_cooldown_remaining() - if cooldown > 0: - self.logger.info( - f"Neighbors: last cycle was too recent, {cooldown:.0f}s left before " - f"another may run" - ) - return self._empty_neighbors_summary( - f"another cycle may run in {cooldown:.0f}s" - ) - - self.neighbors_cycle_active = True try: return await self._run_neighbors_cycle() finally: - self.neighbors_cycle_active = False + self.release_neighbors_cycle() async def _run_neighbors_cycle(self) -> dict[str, Any]: """One discovery cycle: record it, feed the graph, publish it. diff --git a/tests/commands/test_neighbors_command.py b/tests/commands/test_neighbors_command.py index 99fe8a6..4c17d47 100644 --- a/tests/commands/test_neighbors_command.py +++ b/tests/commands/test_neighbors_command.py @@ -33,14 +33,38 @@ def make_service(*, neighbors_enabled=True, summary=None, hang=False, neighbors_cooldown_remaining(service) service.calls = 0 - async def run_cycle(): - service.calls += 1 - if hang: - await asyncio.sleep(30) - if raises is not None: - raise raises - return summary or {"ok": True, "queried": 0, "recorded": 0, "attempted": 0} + def claim_neighbors_cycle(): + if service.neighbors_cycle_active: + return "a discovery cycle is already running" + remaining = service.neighbors_cooldown_remaining() + if remaining > 0: + return f"another cycle may run in {remaining:.0f}s" + service.neighbors_cycle_active = True + return None + def release_neighbors_cycle(): + service.neighbors_cycle_active = False + + async def run_cycle(*, already_claimed=False): + if not already_claimed: + reason = claim_neighbors_cycle() + if reason is not None: + return { + "ok": False, "reason": reason, "discovered": 0, "queried": 0, + "best_snr": None, "attempted": 0, "succeeded": 0, "recorded": 0, + } + service.calls += 1 + try: + if hang: + await asyncio.sleep(30) + if raises is not None: + raise raises + return summary or {"ok": True, "queried": 0, "recorded": 0, "attempted": 0} + finally: + release_neighbors_cycle() + + service.claim_neighbors_cycle = claim_neighbors_cycle + service.release_neighbors_cycle = release_neighbors_cycle service.run_neighbors_cycle = run_cycle return service @@ -55,6 +79,7 @@ def make_command(command_mock_bot, service, *, enabled=True): command._cycle_task = None # Normally set up by BaseCommand.__init__, which make_command skips. command._user_cooldowns = {} + command.cooldown_seconds = NeighborsCommand.cooldown_seconds sent: list[str] = [] @@ -95,6 +120,22 @@ async def test_reports_when_the_service_is_absent(command_mock_bot, message): assert sent == ["disabled"] +async def test_disabled_refusal_does_not_burn_the_senders_own_cooldown( + command_mock_bot, message +): + """Enabling the feature a minute later must not still be blocked for 14 more.""" + command, sent = make_command( + command_mock_bot, make_service(neighbors_enabled=False) + ) + command.record_execution(message.sender_id) + await command.execute(message) + assert sent == ["disabled"] + + can_execute, remaining = command.check_cooldown(message.sender_id) + assert can_execute is False + assert remaining == pytest.approx(60, abs=5) + + async def test_acknowledges_before_running_the_cycle(command_mock_bot, message): """The listen window is ~60s, far too long to hold the reply open.""" service = make_service(summary={"ok": True, "queried": 2, "best_snr": 8.0, @@ -104,10 +145,37 @@ async def test_acknowledges_before_running_the_cycle(command_mock_bot, message): assert await command.execute(message) is True assert sent == ["started seconds=60"] assert service.calls == 0 # not awaited inline + # Claimed before the ack so the scheduler cannot sneak in during send_response. + assert service.neighbors_cycle_active is True await command._cycle_task assert len(sent) == 2 assert sent[1].startswith("success") + assert service.neighbors_cycle_active is False + + +async def test_claim_before_ack_survives_a_scheduler_race(command_mock_bot, message): + """If another trigger starts during send_response, we already hold the lock.""" + service = make_service(summary={"ok": True, "queried": 0, "recorded": 0, "attempted": 0}) + command, sent = make_command(command_mock_bot, service) + + raced = {} + + async def send_and_race(message, content, **kwargs): + sent.append(content) + if content.startswith("started"): + # Stands in for the scheduler waking during the ack await. + raced["summary"] = await service.run_neighbors_cycle() + return True + + command.send_response = send_and_race + await command.execute(message) + await command._cycle_task + + assert sent[0].startswith("started") + assert raced["summary"]["ok"] is False + assert "already running" in raced["summary"]["reason"] + assert service.calls == 1 async def test_summary_reports_count_snr_and_records(command_mock_bot, message): @@ -197,6 +265,22 @@ async def test_a_second_request_will_not_overlap_the_first(command_mock_bot, mes await command._cycle_task +async def test_busy_refusal_does_not_burn_the_senders_own_cooldown( + command_mock_bot, message +): + """A mid-cycle 'busy' clears with the discover window, not after 15 minutes.""" + service = make_service(cycle_active=True) + command, sent = make_command(command_mock_bot, service) + + command.record_execution(message.sender_id) + await command.execute(message) + assert sent == ["busy"] + + can_execute, remaining = command.check_cooldown(message.sender_id) + assert can_execute is False + assert remaining == pytest.approx(60, abs=5) + + async def test_a_finished_cycle_does_not_block_the_next_request(command_mock_bot, message): service = make_service(summary={"ok": True, "queried": 0, "recorded": 0, "attempted": 0}) command, sent = make_command(command_mock_bot, service) diff --git a/tests/unit/test_packet_capture_neighbors.py b/tests/unit/test_packet_capture_neighbors.py index bed45c0..ca13d0d 100644 --- a/tests/unit/test_packet_capture_neighbors.py +++ b/tests/unit/test_packet_capture_neighbors.py @@ -294,6 +294,7 @@ mqtt1_server = one.example.com mqtt1_topic_packets = meshcore/{IATA}/{PUBLIC_KEY}/packets """, radio=FakeRadio()) assert service.global_iata == "xyz" + assert service._iata_is_unset() is True assert service._resolve_neighbors_topic(service.mqtt_brokers[0]) is None @@ -305,10 +306,27 @@ mqtt1_enabled = true mqtt1_server = one.example.com mqtt1_topic_packets = meshcore/{IATA}/{PUBLIC_KEY}/packets """, radio=FakeRadio()) - assert service.global_iata == "xyz" + # Blank stays blank for packet/status topic resolution; neighbors still + # treats it as unset and refuses location-routed publish. + assert service.global_iata == "" + assert service._iata_is_unset() is True assert service._resolve_neighbors_topic(service.mqtt_brokers[0]) is None +def test_empty_iata_keeps_historical_packet_topic_resolution(): + """Empty must not become XYZ for packets — that would publish into XYZ.""" + service = build_service(BASE_INI + """ +iata = +mqtt1_enabled = true +mqtt1_server = one.example.com +mqtt1_topic_packets = meshcore/{IATA}/{PUBLIC_KEY}/packets +""", radio=FakeRadio()) + topic = service._resolve_topic_template( + "meshcore/{IATA}/{PUBLIC_KEY}/packets", "packet" + ) + assert topic == f"meshcore//{SELF_KEY.upper()}/packets" + + def test_a_flat_derived_topic_still_works_without_an_iata(): """The guard is about location routing, not about having an IATA at all.""" service = build_service(BASE_INI + """