refactor(neighbors): improve IATA handling and cycle management

Enhance the handling of the global IATA configuration to ensure that blank values are treated correctly, preventing unintended namespace pollution. Introduce methods for claiming and releasing the neighbors discovery cycle, ensuring that overlapping requests are managed effectively. Update the NeighborsCommand to utilize these new methods, improving the accuracy of cooldown management for users. Additionally, add tests to validate the new behavior and ensure that cooldowns are respected during busy and disabled states.

config.ini.example

@@ -1888,7 +1888,8 @@ iata = XYZ
 #                                     # meshcore/{IATA}/{PUBLIC_KEY}/neighbors), else
 #                                     # <mqttN_topic_prefix>/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)

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(

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.

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)

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 + """