mirror of
https://github.com/pyMC-dev/pyMC_Repeater.git
synced 2026-08-07 17:33:16 +02:00
feat(mqtt): neighbours trigger endpoint, payload counters, persistent schedule
Follow-up work on the neighbours feature, plus one crash found while reviewing it. The crash: `mqtt_brokers.neighbors` is a settings block, but the per-broker key of the same name is a boolean and config.yaml.example documents both, so `neighbors: true` under mqtt_brokers is an easy hand-edit to make. Every reader called .get() on it directly, raising AttributeError inside NeighborScopeHelper.refresh_config -- which is built during daemon init, so the daemon would not start at all. A shared neighbors_config_block() accessor now ignores a non-mapping (with one startup warning instead of per-tick noise) and the API save rewrites it as a proper block. Payload counters, mirroring firmware buildNeighborsMessage: - total_neighbors and queried_neighbors. The latter cannot be derived from `status`: a neighbour the sweep never reached reports `timeout` exactly like one that was asked and stayed silent. ScopeResult therefore carries a `transmitted` flag, set only past the point where the injector confirms the request is on air. - Firmware's third field, `truncated`, is deliberately not emitted -- it reports a fixed PSRAM JSON buffer overflowing, which openhop does not have. total_neighbors here consequently always equals the published row count. - self.default_scope: the region this node stamps on outgoing floods, or `*` when unset. Read from live config so `region default <name>` applies without a restart, and `#`-stripped like the scopes string beside it. Firmware tracks this internally but does not publish it, so this field is openhop-only. POST /api/publish_neighbors runs a cycle immediately, the HTTP twin of the `discover.scopes` mesh CLI command. Authenticated by default along with the rest of /api, and it schedules the cycle on the event loop rather than holding the request open for the minutes a cycle takes. Schedule persistence: _next_publish_at is monotonic and _last_publish_at was in-memory only, so every restart read as "due" and spent a discovery broadcast plus one serialized scope query per neighbour. A generic daemon_state table (migration 14) now records the last publish and start() resumes from it, never sooner than a five-minute grace window and never later than one interval. Two details this had to get right: - The schedule keys off the last *successful* publish. A failed publish also sets _last_publish_at and reschedules on the short retry delay; persisting that would silently turn the retry into a full interval. - _tick's disabled branch no longer clears the schedule before the feature has been enabled in this process. At boot the MQTT connections may not be up, so enabled() can briefly be false, and clearing there would discard the restored schedule and re-run the sweep anyway -- the exact thing this prevents.
This commit is contained in:
agessaman
@@ -562,6 +562,12 @@ mqtt_brokers:
|
||||
# max_neighbors: 32 # cap on neighbours queried per cycle
|
||||
# max_neighbor_age_seconds: 86400 # ignore zero-hop rows older than this
|
||||
# max_sweep_seconds: 900 # give up on a cycle after this long
|
||||
# duty_cycle_abort_seconds: 30 # abandon the sweep when the airtime
|
||||
# # backlog exceeds this
|
||||
#
|
||||
# This must be a block, not a boolean - `neighbors: true` here is ignored (with
|
||||
# a startup warning) because the on/off switch that matters is the per-broker
|
||||
# `neighbors: true` flag documented in the broker schema below.
|
||||
#
|
||||
# `discover.scopes` over the mesh CLI runs one cycle immediately.
|
||||
|
||||
|
||||
@@ -735,11 +735,78 @@ class SQLiteHandler:
|
||||
)
|
||||
logger.info(f"Migration '{migration_name}' applied successfully")
|
||||
|
||||
# Migration 14: Small key/value store for daemon state that must
|
||||
# outlive a restart. Added for the neighbours publisher, whose
|
||||
# schedule otherwise resets on every boot and re-runs a discovery
|
||||
# sweep; kept generic so the next such need does not add another
|
||||
# table. Not for anything hot -- one row per writer, rewritten at
|
||||
# whatever cadence that writer already has.
|
||||
migration_name = "add_daemon_state"
|
||||
existing = conn.execute(
|
||||
"SELECT migration_name FROM migrations WHERE migration_name = ?",
|
||||
(migration_name,),
|
||||
).fetchone()
|
||||
if not existing:
|
||||
conn.execute(
|
||||
"""
|
||||
CREATE TABLE IF NOT EXISTS daemon_state (
|
||||
key TEXT PRIMARY KEY,
|
||||
value_json TEXT NOT NULL,
|
||||
updated_at REAL NOT NULL
|
||||
)
|
||||
"""
|
||||
)
|
||||
conn.execute(
|
||||
"INSERT INTO migrations (migration_name, applied_at) VALUES (?, ?)",
|
||||
(migration_name, time.time()),
|
||||
)
|
||||
logger.info(f"Migration '{migration_name}' applied successfully")
|
||||
|
||||
conn.commit()
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to run migrations: {e}")
|
||||
|
||||
# Daemon state methods
|
||||
def get_daemon_state(self, key: str) -> Optional[dict]:
|
||||
"""Read a persisted daemon-state blob, or None when absent/unreadable.
|
||||
|
||||
Never raises: every caller treats missing state as "no history", so a
|
||||
corrupt row must degrade to that rather than break startup.
|
||||
"""
|
||||
try:
|
||||
with self._connect() as conn:
|
||||
row = conn.execute(
|
||||
"SELECT value_json FROM daemon_state WHERE key = ?", (key,)
|
||||
).fetchone()
|
||||
if not row:
|
||||
return None
|
||||
value = json.loads(row[0])
|
||||
return value if isinstance(value, dict) else None
|
||||
except Exception as e:
|
||||
logger.debug(f"Could not read daemon state '{key}': {e}")
|
||||
return None
|
||||
|
||||
def set_daemon_state(self, key: str, value: dict) -> bool:
|
||||
"""Upsert a daemon-state blob. Returns whether it was written."""
|
||||
try:
|
||||
with self._connect() as conn:
|
||||
conn.execute(
|
||||
"""
|
||||
INSERT INTO daemon_state (key, value_json, updated_at)
|
||||
VALUES (?, ?, ?)
|
||||
ON CONFLICT(key) DO UPDATE SET
|
||||
value_json = excluded.value_json,
|
||||
updated_at = excluded.updated_at
|
||||
""",
|
||||
(key, json.dumps(value), time.time()),
|
||||
)
|
||||
conn.commit()
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.warning(f"Could not persist daemon state '{key}': {e}")
|
||||
return False
|
||||
|
||||
# API Token methods
|
||||
def create_api_token(self, name: str, token_hash: str) -> int:
|
||||
"""Create a new API token entry"""
|
||||
|
||||
@@ -539,6 +539,12 @@ class StorageCollector:
|
||||
def get_neighbors(self) -> dict:
|
||||
return self.sqlite_handler.get_neighbors()
|
||||
|
||||
def get_daemon_state(self, key: str) -> Optional[dict]:
|
||||
return self.sqlite_handler.get_daemon_state(key)
|
||||
|
||||
def set_daemon_state(self, key: str, value: dict) -> bool:
|
||||
return self.sqlite_handler.set_daemon_state(key, value)
|
||||
|
||||
def get_node_name_by_pubkey(self, pubkey: str) -> Optional[str]:
|
||||
"""
|
||||
Lookup node name from adverts table by public key.
|
||||
|
||||
@@ -68,6 +68,32 @@ DEFAULT_MAX_SWEEP_SECONDS = 900.0
|
||||
DEFAULT_DUTY_CYCLE_ABORT_SECONDS = 30.0
|
||||
|
||||
|
||||
def neighbors_config_block(config: Optional[dict]) -> dict:
|
||||
"""Return ``mqtt_brokers.neighbors`` as a mapping, or ``{}`` when it is not one.
|
||||
|
||||
``neighbors`` names a settings block under ``mqtt_brokers`` but a plain
|
||||
boolean on each broker entry, and ``config.yaml.example`` documents both, so
|
||||
``mqtt_brokers.neighbors: true`` is an easy hand-edit to make. A truthy
|
||||
non-mapping used to reach ``.get()`` directly, which raised AttributeError
|
||||
inside :meth:`NeighborScopeHelper.refresh_config` — and because that helper is
|
||||
built during daemon init, it took the whole daemon down on startup. Ignore the
|
||||
value instead; saving from the API rewrites it as a proper block.
|
||||
"""
|
||||
if not isinstance(config, dict):
|
||||
return {}
|
||||
brokers_cfg = config.get("mqtt_brokers", {})
|
||||
if not isinstance(brokers_cfg, dict):
|
||||
return {}
|
||||
block = brokers_cfg.get("neighbors", {})
|
||||
if not isinstance(block, dict):
|
||||
logger.debug(
|
||||
"Ignoring mqtt_brokers.neighbors: expected a settings block, got %s",
|
||||
type(block).__name__,
|
||||
)
|
||||
return {}
|
||||
return block
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class NeighborSnapshot:
|
||||
"""One neighbour, frozen at sweep start.
|
||||
@@ -91,6 +117,12 @@ class NeighborSnapshot:
|
||||
class ScopeResult:
|
||||
status: str
|
||||
scopes: str = ""
|
||||
# Whether the request actually reached the air. Feeds the payload's
|
||||
# ``queried_neighbors``, which firmware increments in ``logTx`` when a QUEUED
|
||||
# entry becomes PENDING. It is not derivable from ``status``: a target the
|
||||
# sweep never reached is reported as ``timeout`` exactly like one that was
|
||||
# asked and stayed silent.
|
||||
transmitted: bool = False
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -149,7 +181,7 @@ class NeighborScopeHelper:
|
||||
config = self.config
|
||||
else:
|
||||
self.config = config
|
||||
neighbors_cfg = (config.get("mqtt_brokers", {}) or {}).get("neighbors", {}) or {}
|
||||
neighbors_cfg = neighbors_config_block(config)
|
||||
|
||||
self._response_timeout_override = _as_float(
|
||||
neighbors_cfg.get("scope_response_timeout_seconds", 0), 0.0
|
||||
@@ -163,8 +195,10 @@ class NeighborScopeHelper:
|
||||
neighbors_cfg.get("duty_cycle_abort_seconds"), DEFAULT_DUTY_CYCLE_ABORT_SECONDS
|
||||
),
|
||||
)
|
||||
delays_cfg = config.get("delays", {}) if isinstance(config, dict) else {}
|
||||
self._direct_tx_delay_factor = _as_float(
|
||||
(config.get("delays", {}) or {}).get("direct_tx_delay_factor"), 0.5
|
||||
delays_cfg.get("direct_tx_delay_factor") if isinstance(delays_cfg, dict) else None,
|
||||
0.5,
|
||||
)
|
||||
|
||||
def response_timeout(self) -> float:
|
||||
@@ -308,13 +342,15 @@ class NeighborScopeHelper:
|
||||
logger.debug(f"Scope request not transmitted for {target.pubkey[:8]}")
|
||||
return ScopeResult(STATUS_SEND_FAILED)
|
||||
|
||||
# Past this point the request is on air, so every outcome counts as
|
||||
# queried regardless of whether the neighbour answers.
|
||||
try:
|
||||
scopes = await asyncio.wait_for(pending.future, timeout)
|
||||
except asyncio.TimeoutError:
|
||||
logger.debug(f"Scope query timed out for {target.pubkey[:8]}")
|
||||
return ScopeResult(STATUS_TIMEOUT)
|
||||
return ScopeResult(STATUS_TIMEOUT, transmitted=True)
|
||||
logger.debug(f"Scope response from {target.pubkey[:8]}: '{scopes}'")
|
||||
return ScopeResult(STATUS_RESPONDED, scopes)
|
||||
return ScopeResult(STATUS_RESPONDED, scopes, transmitted=True)
|
||||
finally:
|
||||
self._pending = None
|
||||
|
||||
|
||||
@@ -31,6 +31,7 @@ from repeater.handler_helpers.neighbor_scopes import (
|
||||
STATUS_TIMEOUT,
|
||||
NeighborSnapshot,
|
||||
ScopeResult,
|
||||
neighbors_config_block,
|
||||
)
|
||||
|
||||
logger = logging.getLogger("NeighborsPublisher")
|
||||
@@ -58,6 +59,19 @@ _TICK_SECONDS = 30.0
|
||||
# waiting out the full interval.
|
||||
RETRY_DELAY_SECONDS = 900.0
|
||||
|
||||
# How an unset default region is spelled in the payload. Matches the wildcard
|
||||
# LoginHelper._format_region_names already emits for unscoped flood.
|
||||
DEFAULT_SCOPE_WILDCARD = "*"
|
||||
|
||||
# daemon_state key holding the persisted schedule.
|
||||
STATE_KEY = "neighbors_publisher"
|
||||
|
||||
# A restored schedule never fires inside this window after boot, even when it is
|
||||
# already overdue. Keeps a restart quiet and lets the radio and brokers settle
|
||||
# before a cycle claims the airtime. Firmware has no equivalent -- it treats
|
||||
# every boot as immediately due.
|
||||
STARTUP_GRACE_SECONDS = 300.0
|
||||
|
||||
# Phases reported by status(), mirroring the firmware's NeighborsPhase.
|
||||
PHASE_DISABLED = "disabled"
|
||||
PHASE_SCHEDULED = "scheduled"
|
||||
@@ -72,6 +86,9 @@ def build_neighbors_payload(
|
||||
self_scopes: str,
|
||||
entries: List[dict],
|
||||
timestamp: Optional[str] = None,
|
||||
total_neighbors: Optional[int] = None,
|
||||
queried_neighbors: Optional[int] = None,
|
||||
self_default_scope: str = DEFAULT_SCOPE_WILDCARD,
|
||||
) -> dict:
|
||||
"""Assemble the ``neighbors`` topic payload.
|
||||
|
||||
@@ -80,18 +97,47 @@ def build_neighbors_payload(
|
||||
needs that order so it can drop the tail when its fixed buffer fills; we keep
|
||||
it because it is the documented shape of the topic and it puts the useful rows
|
||||
first for consumers.
|
||||
|
||||
``self`` carries this node's own advertised scopes plus ``default_scope``, the
|
||||
region it stamps on outgoing floods (``*`` when it floods unscoped). The
|
||||
firmware tracks a ``default_scope`` internally but does not publish it; it is
|
||||
included here because a consumer reading the table cannot otherwise tell which
|
||||
of several scopes this node actually transmits under.
|
||||
|
||||
Two progress counters mirror firmware ``buildNeighborsMessage``:
|
||||
|
||||
* ``total_neighbors`` — neighbours in this cycle's table, which is also how
|
||||
many rows ``neighbors`` carries.
|
||||
* ``queried_neighbors`` — how many scope requests reached the air.
|
||||
|
||||
The firmware's third field, ``truncated``, is deliberately not emitted: it
|
||||
reports that a fixed PSRAM JSON buffer filled and the tail was dropped, and
|
||||
openhop has no such buffer, so it could only ever be false. That also keeps
|
||||
``total_neighbors`` equal to the published row count here, where firmware
|
||||
allows it to run ahead. Both are emitted only when the caller supplies the
|
||||
counts, matching the firmware's ``total_neighbors >= 0`` guard.
|
||||
"""
|
||||
ordered = sorted(
|
||||
entries,
|
||||
key=lambda e: (e.get("heard_secs_ago", 0), -float(e.get("snr", 0.0)), e.get("pubkey", "")),
|
||||
)
|
||||
return {
|
||||
payload = {
|
||||
"timestamp": timestamp or datetime.now(timezone.utc).isoformat(),
|
||||
"origin": origin,
|
||||
"origin_id": origin_id,
|
||||
"self": {"scopes": self_scopes or ""},
|
||||
"neighbors": ordered,
|
||||
}
|
||||
# Key order matches the firmware writer so the two payloads diff cleanly.
|
||||
if total_neighbors is not None:
|
||||
payload["total_neighbors"] = int(total_neighbors)
|
||||
payload["queried_neighbors"] = int(
|
||||
queried_neighbors if queried_neighbors is not None else total_neighbors
|
||||
)
|
||||
payload["self"] = {
|
||||
"scopes": self_scopes or "",
|
||||
"default_scope": self_default_scope or DEFAULT_SCOPE_WILDCARD,
|
||||
}
|
||||
payload["neighbors"] = ordered
|
||||
return payload
|
||||
|
||||
|
||||
class NeighborsPublisher:
|
||||
@@ -124,14 +170,26 @@ class NeighborsPublisher:
|
||||
self._discovery_seen: Dict[str, dict] = {}
|
||||
self._next_publish_at: Optional[float] = None
|
||||
self._last_result: Optional[str] = None
|
||||
# When the last cycle finished, whatever its outcome -- this is what
|
||||
# status() reports alongside last_result.
|
||||
self._last_publish_at: Optional[float] = None
|
||||
# When a cycle last actually reached a broker. The schedule is measured
|
||||
# from this, not from the above: a cycle that failed to publish reschedules
|
||||
# on the short retry delay, and restoring from a failed attempt would
|
||||
# silently turn that retry into a full interval.
|
||||
self._last_success_at: Optional[float] = None
|
||||
# Whether the feature has been enabled at any point in this process. The
|
||||
# disabled branch of _tick clears the schedule so re-enabling publishes
|
||||
# promptly, which must not fire before we have ever been enabled -- that
|
||||
# would throw away a schedule just restored from disk.
|
||||
self._was_enabled = False
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Config accessors (re-read every cycle so live edits take effect)
|
||||
# ------------------------------------------------------------------
|
||||
@property
|
||||
def _neighbors_config(self) -> dict:
|
||||
return (self.config.get("mqtt_brokers", {}) or {}).get("neighbors", {}) or {}
|
||||
return neighbors_config_block(self.config)
|
||||
|
||||
@property
|
||||
def master_enabled(self) -> bool:
|
||||
@@ -181,10 +239,87 @@ class NeighborsPublisher:
|
||||
def start(self) -> None:
|
||||
if self._task is not None and not self._task.done():
|
||||
return
|
||||
# Say so once, loudly, rather than from the config accessors: those run
|
||||
# several times per tick and would bury the log.
|
||||
raw_block = (self.config.get("mqtt_brokers") or {}) if isinstance(self.config, dict) else {}
|
||||
if isinstance(raw_block, dict) and not isinstance(raw_block.get("neighbors", {}), dict):
|
||||
logger.warning(
|
||||
"mqtt_brokers.neighbors is %s, not a settings block - using defaults. "
|
||||
"The per-broker 'neighbors: true' flag is what opts a broker in.",
|
||||
type(raw_block.get("neighbors")).__name__,
|
||||
)
|
||||
self._restore_schedule()
|
||||
self._running = True
|
||||
self._task = asyncio.create_task(self._run_loop(), name="neighbors-publisher")
|
||||
logger.info("Neighbors publisher started (interval %.1fh)", self.interval_seconds / 3600.0)
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Schedule persistence
|
||||
# ------------------------------------------------------------------
|
||||
def _restore_schedule(self) -> None:
|
||||
"""Resume the schedule from the last publish instead of restarting it.
|
||||
|
||||
Without this a restart leaves ``_next_publish_at`` unset, which reads as
|
||||
"due" and spends a discovery broadcast plus a serialized scope query per
|
||||
neighbour on every boot. ``_next_publish_at`` is monotonic and so cannot
|
||||
be stored directly; the persisted value is the wall-clock publish time,
|
||||
converted back to a monotonic deadline here.
|
||||
"""
|
||||
storage = self._storage()
|
||||
reader = getattr(storage, "get_daemon_state", None) if storage else None
|
||||
if not callable(reader):
|
||||
# Older storage backend, or none wired up: behave as before.
|
||||
self._next_publish_at = time.monotonic() + STARTUP_GRACE_SECONDS
|
||||
return
|
||||
|
||||
state = reader(STATE_KEY) or {}
|
||||
last = _as_epoch(state.get("last_success_at"))
|
||||
|
||||
# Restore the display fields too, so a restart does not report the node as
|
||||
# having never run.
|
||||
self._last_result = state.get("last_result") or None
|
||||
self._last_publish_at = _as_epoch(state.get("last_publish_at")) or None
|
||||
interval = self.interval_seconds
|
||||
now = time.time()
|
||||
|
||||
if last <= 0 or last > now:
|
||||
# No successful publish on record, or a timestamp from the future --
|
||||
# the clock moved backwards, or the row is junk. Either way it cannot
|
||||
# place the next cycle, so fall back to the grace delay.
|
||||
if last > now:
|
||||
logger.warning(
|
||||
"Persisted neighbours publish time is %.0fs in the future; ignoring it",
|
||||
last - now,
|
||||
)
|
||||
self._next_publish_at = time.monotonic() + STARTUP_GRACE_SECONDS
|
||||
return
|
||||
|
||||
self._last_success_at = last
|
||||
# Never sooner than the grace window, never later than a full interval
|
||||
# from now -- the latter bounds the damage from an interval that shrank
|
||||
# since the last publish.
|
||||
delay = min(max((last + interval) - now, STARTUP_GRACE_SECONDS), interval)
|
||||
self._next_publish_at = time.monotonic() + delay
|
||||
logger.info(
|
||||
"Neighbours schedule resumed: last published %.1fh ago, next in %.1fh",
|
||||
(now - last) / 3600.0,
|
||||
delay / 3600.0,
|
||||
)
|
||||
|
||||
def _persist_schedule(self) -> None:
|
||||
storage = self._storage()
|
||||
writer = getattr(storage, "set_daemon_state", None) if storage else None
|
||||
if not callable(writer):
|
||||
return
|
||||
writer(
|
||||
STATE_KEY,
|
||||
{
|
||||
"last_success_at": self._last_success_at,
|
||||
"last_publish_at": self._last_publish_at,
|
||||
"last_result": self._last_result,
|
||||
},
|
||||
)
|
||||
|
||||
def trigger_cycle(self) -> bool:
|
||||
"""Start a manual cycle, tracked so shutdown can cancel it.
|
||||
|
||||
@@ -303,10 +438,16 @@ class NeighborsPublisher:
|
||||
async def _tick(self) -> None:
|
||||
if not self.enabled():
|
||||
# Drop the schedule so re-enabling runs a pass promptly, matching the
|
||||
# firmware's next_neighbors_publish = 0 reset.
|
||||
self._next_publish_at = None
|
||||
# firmware's next_neighbors_publish = 0 reset. Only once the feature
|
||||
# has actually been on in this process, though: at boot the MQTT
|
||||
# handler may not have its connections up yet, and clearing here would
|
||||
# discard the schedule just restored from disk and re-run the sweep
|
||||
# anyway -- the exact thing persistence exists to prevent.
|
||||
if self._was_enabled:
|
||||
self._next_publish_at = None
|
||||
return
|
||||
|
||||
self._was_enabled = True
|
||||
handler = self._mqtt_handler()
|
||||
if not handler or not handler.has_connected_neighbors_brokers():
|
||||
logger.debug("Neighbors publish deferred: no connected opted-in broker")
|
||||
@@ -361,6 +502,9 @@ class NeighborsPublisher:
|
||||
else "publish failed (broker unreachable or rejected the payload)"
|
||||
)
|
||||
self._last_publish_at = time.time()
|
||||
if published:
|
||||
self._last_success_at = self._last_publish_at
|
||||
self._persist_schedule()
|
||||
logger.info(
|
||||
"Neighbors %s cycle finished in %.1fs: %d neighbour(s), %d with scopes, "
|
||||
"published=%s",
|
||||
@@ -486,6 +630,8 @@ class NeighborsPublisher:
|
||||
# the same ordering the firmware applies before it starts querying.
|
||||
snapshots.sort(key=lambda s: (-s.last_seen, -s.snr, s.pubkey))
|
||||
if len(snapshots) > self.max_neighbors:
|
||||
# Only logged, not published: the payload reports the capped table, so
|
||||
# the dropped rows would otherwise be invisible here.
|
||||
logger.info(
|
||||
"Neighbour table has %d entries; querying the freshest %d",
|
||||
len(snapshots),
|
||||
@@ -495,12 +641,17 @@ class NeighborsPublisher:
|
||||
return snapshots
|
||||
|
||||
def _build_payload(
|
||||
self, targets: List[NeighborSnapshot], scope_results: Dict[str, ScopeResult]
|
||||
self,
|
||||
targets: List[NeighborSnapshot],
|
||||
scope_results: Dict[str, ScopeResult],
|
||||
) -> dict:
|
||||
now = time.time()
|
||||
entries = []
|
||||
queried = 0
|
||||
for target in targets:
|
||||
result = scope_results.get(target.pubkey) or ScopeResult(STATUS_TIMEOUT)
|
||||
if result.transmitted:
|
||||
queried += 1
|
||||
heard_secs_ago = int(max(0.0, now - target.last_seen)) if target.last_seen else 0
|
||||
entries.append(
|
||||
{
|
||||
@@ -527,9 +678,33 @@ class NeighborsPublisher:
|
||||
origin=origin,
|
||||
origin_id=origin_id,
|
||||
self_scopes=self._self_scopes(),
|
||||
self_default_scope=self._self_default_scope(),
|
||||
entries=entries,
|
||||
total_neighbors=len(entries),
|
||||
queried_neighbors=queried,
|
||||
)
|
||||
|
||||
def _self_default_scope(self) -> str:
|
||||
"""The region this node stamps on outgoing floods, or ``*`` when unset.
|
||||
|
||||
Read from live config on every cycle because ``region default <name>`` over
|
||||
the mesh CLI writes straight into ``config["mesh"]`` (the same dict this
|
||||
holds) and expects to take effect without a restart.
|
||||
|
||||
Normalised like the ``scopes`` field beside it: the transport-key table
|
||||
stores region names with a leading ``#``, which is not part of the name a
|
||||
consumer matches on, so it is stripped here as
|
||||
``LoginHelper._format_region_names`` strips it there.
|
||||
"""
|
||||
mesh_cfg = self.config.get("mesh", {}) if isinstance(self.config, dict) else {}
|
||||
if not isinstance(mesh_cfg, dict):
|
||||
return DEFAULT_SCOPE_WILDCARD
|
||||
raw = mesh_cfg.get("default_region")
|
||||
name = str(raw).strip() if raw not in (None, "") else ""
|
||||
if name.startswith("#"):
|
||||
name = name[1:].strip()
|
||||
return name or DEFAULT_SCOPE_WILDCARD
|
||||
|
||||
def _self_scopes(self) -> str:
|
||||
if not self._self_scopes_fn:
|
||||
return ""
|
||||
@@ -551,6 +726,16 @@ class NeighborsPublisher:
|
||||
return bool(results)
|
||||
|
||||
|
||||
def _as_epoch(value) -> float:
|
||||
"""Coerce a persisted timestamp to a float, or 0.0 when it is unusable."""
|
||||
try:
|
||||
if value is None:
|
||||
return 0.0
|
||||
return float(value)
|
||||
except (TypeError, ValueError):
|
||||
return 0.0
|
||||
|
||||
|
||||
def normalize_interval_hours(value) -> float:
|
||||
"""Validate an interval in hours, falling back to the default when invalid.
|
||||
|
||||
|
||||
@@ -2337,6 +2337,60 @@ class APIEndpoints:
|
||||
logger.error(f"Error listing broker presets: {e}")
|
||||
return self._error(str(e))
|
||||
|
||||
@cherrypy.expose
|
||||
@cherrypy.tools.json_out()
|
||||
def publish_neighbors(self):
|
||||
"""Run one neighbours discovery + publish cycle now.
|
||||
|
||||
POST /api/publish_neighbors
|
||||
|
||||
The HTTP equivalent of the ``discover.scopes`` mesh CLI command. A cycle
|
||||
runs for minutes -- a discovery window plus one serialized scope query per
|
||||
neighbour -- so this schedules it on the event loop and returns
|
||||
immediately rather than holding the request open. Poll ``mqtt_status``
|
||||
for the outcome.
|
||||
"""
|
||||
self._set_cors_headers()
|
||||
|
||||
if cherrypy.request.method == "OPTIONS":
|
||||
return ""
|
||||
|
||||
future = None
|
||||
try:
|
||||
self._require_post()
|
||||
publisher = getattr(self.daemon_instance, "neighbors_publisher", None)
|
||||
if not publisher:
|
||||
return self._error("Neighbors publisher not available")
|
||||
if not publisher.enabled():
|
||||
return self._error(
|
||||
"Neighbours publishing is disabled - enable it and opt a broker in first"
|
||||
)
|
||||
if self.event_loop is None:
|
||||
return self._error("Event loop not available")
|
||||
|
||||
import asyncio
|
||||
|
||||
# trigger_cycle owns the already-running check and tracks the task so
|
||||
# shutdown can cancel it. Doing either here would race: this runs on a
|
||||
# cherrypy thread while the cycle lives on the event loop.
|
||||
async def _start():
|
||||
return publisher.trigger_cycle()
|
||||
|
||||
future = asyncio.run_coroutine_threadsafe(_start(), self.event_loop)
|
||||
if future.result(timeout=10):
|
||||
return self._success("Neighbours discovery cycle started")
|
||||
return self._error("A neighbours cycle is already running")
|
||||
except FutureTimeoutError:
|
||||
logger.error("Timed out starting the neighbours cycle", exc_info=True)
|
||||
if future is not None:
|
||||
future.cancel()
|
||||
return self._error("Timed out starting the neighbours cycle")
|
||||
except cherrypy.HTTPError:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Error starting neighbours cycle: {e}", exc_info=True)
|
||||
return self._error(str(e))
|
||||
|
||||
@staticmethod
|
||||
def _validate_neighbors_settings(raw):
|
||||
"""Validate the ``mqtt_brokers.neighbors`` block.
|
||||
@@ -2489,16 +2543,22 @@ class APIEndpoints:
|
||||
if "email" in data:
|
||||
mqtt_updates["email"] = str(data["email"]).strip()
|
||||
if "neighbors" in data:
|
||||
from repeater.handler_helpers.neighbor_scopes import neighbors_config_block
|
||||
|
||||
neighbors_settings, error = self._validate_neighbors_settings(data["neighbors"])
|
||||
if error:
|
||||
return self._error(error)
|
||||
# update_and_save replaces a section's key outright, so merge onto
|
||||
# the stored block instead of letting a partial POST drop the
|
||||
# settings it did not mention.
|
||||
existing_neighbors = (self.config.get("mqtt_brokers", {}) or {}).get(
|
||||
"neighbors", {}
|
||||
) or {}
|
||||
mqtt_updates["neighbors"] = {**existing_neighbors, **neighbors_settings}
|
||||
# settings it did not mention. neighbors_config_block returns {} for
|
||||
# a hand-edited scalar (`neighbors: true` is an easy mistake, since
|
||||
# the per-broker key of the same name is a boolean), which would
|
||||
# otherwise fail the merge with a TypeError; the save then rewrites
|
||||
# it as a proper block.
|
||||
mqtt_updates["neighbors"] = {
|
||||
**neighbors_config_block(self.config),
|
||||
**neighbors_settings,
|
||||
}
|
||||
# if "disallowed_packet_types" in data:
|
||||
# mqtt_updates["disallowed_packet_types"] = list(data["disallowed_packet_types"])
|
||||
if "brokers" in data:
|
||||
|
||||
@@ -3417,6 +3417,28 @@ paths:
|
||||
schema:
|
||||
type: object
|
||||
|
||||
/publish_neighbors:
|
||||
post:
|
||||
tags: [System]
|
||||
summary: Publish the neighbours table now
|
||||
description: >
|
||||
Runs one neighbours cycle immediately: a zero-hop discovery broadcast, a
|
||||
serialized scope query per neighbour, then a publish to every opted-in
|
||||
broker. Returns as soon as the cycle is scheduled; the cycle itself takes
|
||||
minutes. Poll /mqtt_status for the outcome.
|
||||
security:
|
||||
- BearerAuth: []
|
||||
- ApiKeyAuth: []
|
||||
responses:
|
||||
'200':
|
||||
description: Cycle started, or an error when one is already running
|
||||
content:
|
||||
application/json:
|
||||
schema:
|
||||
$ref: '#/components/schemas/SuccessResponse'
|
||||
'405':
|
||||
description: Method not allowed
|
||||
|
||||
/update_web_config:
|
||||
post:
|
||||
tags: [System]
|
||||
|
||||
@@ -21,6 +21,7 @@ from openhop_core.protocol import CryptoUtils, Identity, LocalIdentity
|
||||
from openhop_core.protocol.constants import PAYLOAD_TYPE_RESPONSE
|
||||
from repeater.data_acquisition.mqtt_handler import MeshCoreToMqttPusher
|
||||
from repeater.handler_helpers.neighbor_scopes import (
|
||||
DEFAULT_MAX_SWEEP_SECONDS,
|
||||
STATUS_RESPONDED,
|
||||
STATUS_SEND_FAILED,
|
||||
STATUS_TIMEOUT,
|
||||
@@ -28,8 +29,11 @@ from repeater.handler_helpers.neighbor_scopes import (
|
||||
NeighborSnapshot,
|
||||
)
|
||||
from repeater.neighbors_publisher import (
|
||||
DEFAULT_INTERVAL_HOURS,
|
||||
MAX_INTERVAL_HOURS,
|
||||
MIN_INTERVAL_HOURS,
|
||||
STARTUP_GRACE_SECONDS,
|
||||
STATE_KEY,
|
||||
NeighborsPublisher,
|
||||
build_neighbors_payload,
|
||||
normalize_interval_hours,
|
||||
@@ -100,7 +104,7 @@ def test_payload_orders_most_useful_first():
|
||||
)
|
||||
|
||||
assert [e["pubkey"] for e in payload["neighbors"]] == ["bb", "aa", "cc"]
|
||||
assert payload["self"] == {"scopes": "DEN,APRS"}
|
||||
assert payload["self"] == {"scopes": "DEN,APRS", "default_scope": "*"}
|
||||
assert payload["origin_id"] == "AA" * 32
|
||||
assert payload["timestamp"]
|
||||
|
||||
@@ -823,7 +827,7 @@ async def test_cycle_publishes_table_with_unanswered_neighbors():
|
||||
assert result["responded"] == 1
|
||||
|
||||
payload = published[0]
|
||||
assert payload["self"] == {"scopes": "DEN,APRS"}
|
||||
assert payload["self"] == {"scopes": "DEN,APRS", "default_scope": "*"}
|
||||
statuses = {e["pubkey"]: e["status"] for e in payload["neighbors"]}
|
||||
assert statuses == {"aa" * 32: STATUS_RESPONDED, "bb" * 32: STATUS_TIMEOUT}
|
||||
# A cycle always arms the next one, so a failure cannot wedge the schedule.
|
||||
@@ -1015,3 +1019,576 @@ def test_invalid_neighbors_interval_is_rejected_by_the_endpoint(monkeypatch):
|
||||
assert out["success"] is False
|
||||
assert "between 12 and 336" in out["error"]
|
||||
api.config_manager.update_and_save.assert_not_called()
|
||||
|
||||
|
||||
# ====================================================================
|
||||
# Malformed config block
|
||||
# ====================================================================
|
||||
# `neighbors` is a settings block under mqtt_brokers but a boolean on each broker
|
||||
# entry, and config.yaml.example documents both, so `mqtt_brokers.neighbors: true`
|
||||
# is an easy hand-edit to make. Every reader used to call .get() on it directly.
|
||||
@pytest.mark.parametrize("bogus", [True, 24, "on", ["DEN"]])
|
||||
def test_scalar_neighbors_block_is_ignored_not_fatal(bogus):
|
||||
"""A truthy scalar took the daemon down: the scope helper reads it in __init__."""
|
||||
config = {"mqtt_brokers": {"neighbors": bogus}}
|
||||
|
||||
helper = NeighborScopeHelper(
|
||||
local_identity=LocalIdentity(), packet_injector=None, config=config
|
||||
)
|
||||
assert helper._max_sweep_seconds == DEFAULT_MAX_SWEEP_SECONDS
|
||||
|
||||
publisher = NeighborsPublisher(config=config)
|
||||
assert publisher._neighbors_config == {}
|
||||
assert publisher.master_enabled is True
|
||||
assert publisher.interval_seconds == DEFAULT_INTERVAL_HOURS * 3600.0
|
||||
assert publisher.enabled() is False # no handler, so nothing is published
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bogus", [True, 24, "on"])
|
||||
def test_scalar_neighbors_block_is_repaired_by_a_save(monkeypatch, bogus):
|
||||
"""The merge onto the stored block must not fail on a non-mapping."""
|
||||
api, request = _api_with_stored_brokers(monkeypatch, [], neighbors_block=bogus)
|
||||
|
||||
request.json = {"neighbors": {"enabled": True, "interval_hours": 36}}
|
||||
assert api.update_mqtt_config()["success"] is True
|
||||
|
||||
saved = api.config_manager.update_and_save.call_args.kwargs["updates"]["mqtt_brokers"]
|
||||
assert saved["neighbors"] == {"enabled": True, "interval_hours": 36}
|
||||
|
||||
|
||||
def test_scalar_delays_block_does_not_break_the_response_window():
|
||||
helper = NeighborScopeHelper(
|
||||
local_identity=LocalIdentity(),
|
||||
packet_injector=None,
|
||||
config={"delays": True},
|
||||
)
|
||||
assert helper._direct_tx_delay_factor == 0.5
|
||||
|
||||
|
||||
# ====================================================================
|
||||
# Progress metadata (firmware total_neighbors / queried_neighbors)
|
||||
# ====================================================================
|
||||
def test_payload_reports_progress_metadata_in_firmware_key_order():
|
||||
payload = build_neighbors_payload(
|
||||
origin="node",
|
||||
origin_id="AA" * 32,
|
||||
self_scopes="DEN",
|
||||
entries=[
|
||||
{
|
||||
"pubkey": "aa",
|
||||
"snr": 1.0,
|
||||
"heard_secs_ago": 5,
|
||||
"scopes": "DEN",
|
||||
"status": "responded",
|
||||
}
|
||||
],
|
||||
total_neighbors=4,
|
||||
queried_neighbors=2,
|
||||
)
|
||||
|
||||
assert payload["total_neighbors"] == 4
|
||||
assert payload["queried_neighbors"] == 2
|
||||
# Firmware's buildNeighborsMessageBase writes the counters between origin_id
|
||||
# and self; keeping the order lets the two payloads diff cleanly.
|
||||
assert list(payload) == [
|
||||
"timestamp",
|
||||
"origin",
|
||||
"origin_id",
|
||||
"total_neighbors",
|
||||
"queried_neighbors",
|
||||
"self",
|
||||
"neighbors",
|
||||
]
|
||||
|
||||
|
||||
def test_payload_never_emits_the_firmware_truncated_field():
|
||||
"""It reports a fixed PSRAM buffer overflowing, which openhop cannot have."""
|
||||
payload = build_neighbors_payload(
|
||||
origin="node",
|
||||
origin_id="AA" * 32,
|
||||
self_scopes="",
|
||||
entries=[],
|
||||
total_neighbors=3,
|
||||
queried_neighbors=1,
|
||||
)
|
||||
|
||||
assert "truncated" not in payload
|
||||
|
||||
|
||||
def test_payload_omits_progress_metadata_when_counts_are_absent():
|
||||
"""Mirrors the firmware's `total_neighbors >= 0` guard."""
|
||||
payload = build_neighbors_payload(
|
||||
origin="node", origin_id="AA" * 32, self_scopes="", entries=[]
|
||||
)
|
||||
|
||||
assert "total_neighbors" not in payload
|
||||
assert "queried_neighbors" not in payload
|
||||
|
||||
|
||||
def test_queried_count_excludes_neighbors_never_put_on_air():
|
||||
"""`status` cannot carry this: an unreached target also reports `timeout`."""
|
||||
from repeater.handler_helpers.neighbor_scopes import ScopeResult
|
||||
|
||||
targets = [
|
||||
NeighborSnapshot(pubkey=f"{i:064x}", last_seen=time.time(), snr=1.0) for i in range(4)
|
||||
]
|
||||
publisher = _publisher({"mqtt_brokers": {}})
|
||||
|
||||
payload = publisher._build_payload(
|
||||
targets,
|
||||
{
|
||||
targets[0].pubkey: ScopeResult(STATUS_RESPONDED, "DEN", transmitted=True),
|
||||
targets[1].pubkey: ScopeResult(STATUS_TIMEOUT, transmitted=True),
|
||||
targets[2].pubkey: ScopeResult(STATUS_SEND_FAILED), # never transmitted
|
||||
targets[3].pubkey: ScopeResult(STATUS_TIMEOUT), # sweep never reached it
|
||||
},
|
||||
)
|
||||
|
||||
assert payload["queried_neighbors"] == 2
|
||||
|
||||
|
||||
def test_total_neighbors_always_matches_the_published_row_count():
|
||||
"""Without `truncated` to flag a gap, the two must not diverge."""
|
||||
publisher = _publisher({"mqtt_brokers": {}})
|
||||
targets = [
|
||||
NeighborSnapshot(pubkey=f"{i:064x}", last_seen=time.time(), snr=1.0) for i in range(3)
|
||||
]
|
||||
|
||||
payload = publisher._build_payload(targets, {})
|
||||
|
||||
assert payload["total_neighbors"] == len(payload["neighbors"]) == 3
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_transmitted_flag_tracks_whether_the_request_reached_the_air():
|
||||
local = LocalIdentity()
|
||||
peer = LocalIdentity()
|
||||
target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time())
|
||||
|
||||
async def failed(packet, wait_for_ack=False):
|
||||
return False
|
||||
|
||||
results = await _helper_with_injector(local, failed).sweep([target])
|
||||
assert results[target.pubkey].transmitted is False
|
||||
|
||||
async def sent(packet, wait_for_ack=False):
|
||||
return True
|
||||
|
||||
helper = _helper_with_injector(
|
||||
local,
|
||||
sent,
|
||||
config={"mqtt_brokers": {"neighbors": {"scope_response_timeout_seconds": 0.05}}},
|
||||
)
|
||||
results = await helper.sweep([target])
|
||||
assert results[target.pubkey].status == STATUS_TIMEOUT
|
||||
assert results[target.pubkey].transmitted is True
|
||||
|
||||
|
||||
# ====================================================================
|
||||
# Manual trigger endpoint
|
||||
# ====================================================================
|
||||
def _api_with_publisher(monkeypatch, publisher, method="POST"):
|
||||
import cherrypy
|
||||
|
||||
from repeater.web.api_endpoints import APIEndpoints
|
||||
|
||||
request = SimpleNamespace(method=method, params={}, json={})
|
||||
response = SimpleNamespace(headers={}, status=200)
|
||||
monkeypatch.setattr(cherrypy, "request", request, raising=False)
|
||||
monkeypatch.setattr(cherrypy, "response", response, raising=False)
|
||||
|
||||
api = APIEndpoints.__new__(APIEndpoints)
|
||||
api.config = {}
|
||||
api.daemon_instance = SimpleNamespace(neighbors_publisher=publisher)
|
||||
api.event_loop = asyncio.new_event_loop()
|
||||
api.send_advert_func = None
|
||||
api.stats_getter = None
|
||||
api._config_path = "/tmp/test-config.yaml"
|
||||
api.config_manager = MagicMock()
|
||||
return api
|
||||
|
||||
|
||||
class _FakePublisher:
|
||||
def __init__(self, *, is_enabled=True, starts=True):
|
||||
self._enabled = is_enabled
|
||||
self._starts = starts
|
||||
self.triggered = 0
|
||||
|
||||
def enabled(self):
|
||||
return self._enabled
|
||||
|
||||
def trigger_cycle(self):
|
||||
self.triggered += 1
|
||||
return self._starts
|
||||
|
||||
|
||||
def _run_endpoint(api):
|
||||
"""Drive the endpoint's run_coroutine_threadsafe against a real loop."""
|
||||
import threading
|
||||
|
||||
loop = api.event_loop
|
||||
thread = threading.Thread(target=loop.run_forever, daemon=True)
|
||||
thread.start()
|
||||
try:
|
||||
return api.publish_neighbors()
|
||||
finally:
|
||||
loop.call_soon_threadsafe(loop.stop)
|
||||
thread.join(timeout=5)
|
||||
# Closing a loop whose thread has not finished unwinding can raise on its
|
||||
# self-pipe descriptors; leak it rather than risk a flaky teardown.
|
||||
if not thread.is_alive():
|
||||
loop.close()
|
||||
|
||||
|
||||
def test_publish_neighbors_endpoint_starts_a_cycle(monkeypatch):
|
||||
publisher = _FakePublisher()
|
||||
api = _api_with_publisher(monkeypatch, publisher)
|
||||
|
||||
out = _run_endpoint(api)
|
||||
|
||||
assert out["success"] is True
|
||||
assert publisher.triggered == 1
|
||||
|
||||
|
||||
def test_publish_neighbors_endpoint_reports_an_already_running_cycle(monkeypatch):
|
||||
publisher = _FakePublisher(starts=False)
|
||||
api = _api_with_publisher(monkeypatch, publisher)
|
||||
|
||||
out = _run_endpoint(api)
|
||||
|
||||
assert out["success"] is False
|
||||
assert "already running" in out["error"]
|
||||
|
||||
|
||||
def test_publish_neighbors_endpoint_refuses_when_disabled(monkeypatch):
|
||||
"""No broker opted in: refuse rather than burn airtime on an unpublishable cycle."""
|
||||
publisher = _FakePublisher(is_enabled=False)
|
||||
api = _api_with_publisher(monkeypatch, publisher)
|
||||
|
||||
out = api.publish_neighbors()
|
||||
|
||||
assert out["success"] is False
|
||||
assert publisher.triggered == 0
|
||||
|
||||
|
||||
def test_publish_neighbors_endpoint_without_a_publisher(monkeypatch):
|
||||
api = _api_with_publisher(monkeypatch, None)
|
||||
|
||||
out = api.publish_neighbors()
|
||||
|
||||
assert out["success"] is False
|
||||
assert "not available" in out["error"]
|
||||
|
||||
|
||||
# ====================================================================
|
||||
# self.default_scope
|
||||
# ====================================================================
|
||||
@pytest.mark.parametrize(
|
||||
"mesh_cfg,expected",
|
||||
[
|
||||
({"default_region": "DEN"}, "DEN"),
|
||||
({"default_region": "#DEN"}, "DEN"), # transport-key tables prefix with '#'
|
||||
({"default_region": " DEN "}, "DEN"),
|
||||
({"default_region": None}, "*"), # unset -> floods unscoped
|
||||
({"default_region": ""}, "*"),
|
||||
({"default_region": " "}, "*"),
|
||||
({}, "*"), # key absent entirely
|
||||
({"default_region": "*"}, "*"),
|
||||
(True, "*"), # hand-edited scalar must not raise
|
||||
],
|
||||
)
|
||||
def test_self_default_scope_normalisation(mesh_cfg, expected):
|
||||
publisher = _publisher({"mqtt_brokers": {}, "mesh": mesh_cfg})
|
||||
assert publisher._self_default_scope() == expected
|
||||
|
||||
|
||||
def test_default_scope_is_published_inside_self():
|
||||
publisher = _publisher({"mqtt_brokers": {}, "mesh": {"default_region": "#PDX"}})
|
||||
|
||||
payload = publisher._build_payload([], {})
|
||||
|
||||
assert payload["self"]["default_scope"] == "PDX"
|
||||
assert "scopes" in payload["self"]
|
||||
|
||||
|
||||
def test_default_scope_tracks_a_live_config_edit():
|
||||
"""`region default <name>` writes into the same dict and must not need a restart."""
|
||||
config = {"mqtt_brokers": {}, "mesh": {"default_region": None}}
|
||||
publisher = _publisher(config)
|
||||
|
||||
assert publisher._build_payload([], {})["self"]["default_scope"] == "*"
|
||||
|
||||
config["mesh"]["default_region"] = "DEN"
|
||||
assert publisher._build_payload([], {})["self"]["default_scope"] == "DEN"
|
||||
|
||||
|
||||
def test_payload_defaults_default_scope_to_the_wildcard():
|
||||
"""Callers that omit it still emit a valid self block."""
|
||||
payload = build_neighbors_payload(
|
||||
origin="node", origin_id="AA" * 32, self_scopes="DEN", entries=[]
|
||||
)
|
||||
|
||||
assert payload["self"] == {"scopes": "DEN", "default_scope": "*"}
|
||||
|
||||
|
||||
# ====================================================================
|
||||
# Schedule persistence across restarts
|
||||
# ====================================================================
|
||||
class _FakeStateStore:
|
||||
"""Stands in for the daemon_state table."""
|
||||
|
||||
def __init__(self, initial=None):
|
||||
self.rows = dict(initial or {})
|
||||
self.writes = 0
|
||||
|
||||
def get_daemon_state(self, key):
|
||||
return self.rows.get(key)
|
||||
|
||||
def set_daemon_state(self, key, value):
|
||||
self.rows[key] = dict(value)
|
||||
self.writes += 1
|
||||
return True
|
||||
|
||||
|
||||
def _enabled_handler():
|
||||
return SimpleNamespace(
|
||||
has_neighbors_brokers=lambda: True, has_connected_neighbors_brokers=lambda: True
|
||||
)
|
||||
|
||||
|
||||
def test_restore_resumes_the_interval_from_the_last_successful_publish():
|
||||
now = time.time()
|
||||
store = _FakeStateStore({STATE_KEY: {"last_success_at": now - 3600, "last_result": "ok"}})
|
||||
publisher = _publisher(
|
||||
{"mqtt_brokers": {"neighbors": {"interval_hours": 24}}},
|
||||
handler=_enabled_handler(),
|
||||
storage=store,
|
||||
)
|
||||
|
||||
publisher._restore_schedule()
|
||||
|
||||
# 1h since the last publish on a 24h interval -> ~23h to go, not "due now".
|
||||
secs = publisher.status()["secs_until_next"]
|
||||
assert 22.9 * 3600 < secs < 23.1 * 3600
|
||||
assert publisher.status()["phase"] == "scheduled"
|
||||
assert publisher._last_result == "ok"
|
||||
|
||||
|
||||
def test_restore_applies_the_grace_delay_when_already_overdue():
|
||||
"""A node off for a week must not transmit the instant it boots."""
|
||||
store = _FakeStateStore(
|
||||
{STATE_KEY: {"last_success_at": time.time() - 7 * 86400, "last_result": "ok"}}
|
||||
)
|
||||
publisher = _publisher(
|
||||
{"mqtt_brokers": {"neighbors": {"interval_hours": 24}}},
|
||||
handler=_enabled_handler(),
|
||||
storage=store,
|
||||
)
|
||||
|
||||
publisher._restore_schedule()
|
||||
|
||||
assert publisher.status()["secs_until_next"] == pytest.approx(STARTUP_GRACE_SECONDS, abs=2)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"state",
|
||||
[
|
||||
None, # fresh install
|
||||
{},
|
||||
{"last_success_at": None},
|
||||
{"last_success_at": "nonsense"},
|
||||
{"last_success_at": 0},
|
||||
# Clock moved backwards, or a junk row: must not park the schedule in the
|
||||
# far future where nothing would ever publish again.
|
||||
{"last_success_at": time.time() + 5 * 86400},
|
||||
],
|
||||
)
|
||||
def test_restore_falls_back_to_the_grace_delay_on_unusable_state(state):
|
||||
store = _FakeStateStore({STATE_KEY: state} if state is not None else {})
|
||||
publisher = _publisher(
|
||||
{"mqtt_brokers": {"neighbors": {"interval_hours": 24}}},
|
||||
handler=_enabled_handler(),
|
||||
storage=store,
|
||||
)
|
||||
|
||||
publisher._restore_schedule()
|
||||
|
||||
assert publisher.status()["secs_until_next"] == pytest.approx(STARTUP_GRACE_SECONDS, abs=2)
|
||||
|
||||
|
||||
def test_restore_clamps_a_delay_longer_than_the_interval():
|
||||
"""An interval shortened since the last publish must take effect now."""
|
||||
store = _FakeStateStore({STATE_KEY: {"last_success_at": time.time()}})
|
||||
publisher = _publisher(
|
||||
{"mqtt_brokers": {"neighbors": {"interval_hours": 12}}},
|
||||
handler=_enabled_handler(),
|
||||
storage=store,
|
||||
)
|
||||
publisher._restore_schedule()
|
||||
assert publisher.status()["secs_until_next"] <= 12 * 3600
|
||||
|
||||
# Same stored publish time, but the config now says 24h -> still bounded.
|
||||
publisher.config["mqtt_brokers"]["neighbors"]["interval_hours"] = 24
|
||||
publisher._restore_schedule()
|
||||
assert publisher.status()["secs_until_next"] <= 24 * 3600
|
||||
|
||||
|
||||
def test_restore_without_a_state_capable_storage_backend():
|
||||
"""An older storage backend must still start, just without resuming."""
|
||||
publisher = _publisher({"mqtt_brokers": {}}, handler=_enabled_handler(), storage=object())
|
||||
|
||||
publisher._restore_schedule()
|
||||
|
||||
assert publisher.status()["secs_until_next"] == pytest.approx(STARTUP_GRACE_SECONDS, abs=2)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_failed_publish_is_not_recorded_as_a_successful_one():
|
||||
"""Otherwise a restart turns the 15-minute retry into a full interval."""
|
||||
store = _FakeStateStore()
|
||||
handler = SimpleNamespace(
|
||||
has_neighbors_brokers=lambda: True,
|
||||
has_connected_neighbors_brokers=lambda: True,
|
||||
publish_neighbors=lambda payload: [], # reached no broker
|
||||
node_name="n",
|
||||
public_key="AB" * 32,
|
||||
)
|
||||
store.get_neighbors = lambda: {}
|
||||
publisher = _publisher({"mqtt_brokers": {}}, handler=handler, storage=store)
|
||||
|
||||
await publisher.run_cycle(trigger="test")
|
||||
|
||||
saved = store.rows[STATE_KEY]
|
||||
assert saved["last_success_at"] is None
|
||||
assert saved["last_publish_at"] is not None # the attempt is still recorded
|
||||
assert "publish failed" in saved["last_result"]
|
||||
|
||||
# A restart therefore retries promptly rather than waiting out the interval.
|
||||
restarted = _publisher(
|
||||
{"mqtt_brokers": {"neighbors": {"interval_hours": 24}}},
|
||||
handler=_enabled_handler(),
|
||||
storage=store,
|
||||
)
|
||||
restarted._restore_schedule()
|
||||
assert restarted.status()["secs_until_next"] == pytest.approx(STARTUP_GRACE_SECONDS, abs=2)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_successful_publish_persists_a_resumable_schedule():
|
||||
store = _FakeStateStore()
|
||||
handler = SimpleNamespace(
|
||||
has_neighbors_brokers=lambda: True,
|
||||
has_connected_neighbors_brokers=lambda: True,
|
||||
publish_neighbors=lambda payload: [("broker", None)],
|
||||
node_name="n",
|
||||
public_key="AB" * 32,
|
||||
)
|
||||
store.get_neighbors = lambda: {}
|
||||
publisher = _publisher({"mqtt_brokers": {}}, handler=handler, storage=store)
|
||||
|
||||
await publisher.run_cycle(trigger="test")
|
||||
|
||||
assert store.rows[STATE_KEY]["last_success_at"] == pytest.approx(time.time(), abs=5)
|
||||
|
||||
restarted = _publisher(
|
||||
{"mqtt_brokers": {"neighbors": {"interval_hours": 24}}},
|
||||
handler=_enabled_handler(),
|
||||
storage=store,
|
||||
)
|
||||
restarted._restore_schedule()
|
||||
assert restarted.status()["secs_until_next"] > 23 * 3600
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_boot_tick_does_not_discard_the_restored_schedule():
|
||||
"""The disabled branch of _tick must not fire before we were ever enabled.
|
||||
|
||||
At boot the MQTT connections may not be up, so enabled() can briefly be
|
||||
False. Clearing the schedule there would mark the node due and re-run the
|
||||
sweep on every restart -- the exact thing persistence prevents.
|
||||
"""
|
||||
store = _FakeStateStore({STATE_KEY: {"last_success_at": time.time() - 3600}})
|
||||
not_ready = SimpleNamespace(has_neighbors_brokers=lambda: False)
|
||||
publisher = _publisher(
|
||||
{"mqtt_brokers": {"neighbors": {"interval_hours": 24}}},
|
||||
handler=not_ready,
|
||||
storage=store,
|
||||
)
|
||||
publisher._restore_schedule()
|
||||
restored = publisher._next_publish_at
|
||||
|
||||
await publisher._tick()
|
||||
|
||||
assert publisher._next_publish_at == restored
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_disabling_after_being_enabled_still_clears_the_schedule():
|
||||
"""Re-enabling should publish promptly; that behaviour is preserved."""
|
||||
store = _FakeStateStore()
|
||||
enabled = True
|
||||
handler = SimpleNamespace(
|
||||
has_neighbors_brokers=lambda: enabled,
|
||||
has_connected_neighbors_brokers=lambda: False,
|
||||
)
|
||||
publisher = _publisher({"mqtt_brokers": {}}, handler=handler, storage=store)
|
||||
publisher._next_publish_at = time.monotonic() + 3600
|
||||
|
||||
await publisher._tick() # enabled, but no connected broker -> schedule kept
|
||||
assert publisher._next_publish_at is not None
|
||||
|
||||
enabled = False
|
||||
await publisher._tick()
|
||||
assert publisher._next_publish_at is None
|
||||
|
||||
|
||||
def test_daemon_state_round_trips_through_real_sqlite(tmp_path):
|
||||
"""The fake store above cannot prove the migration or the accessors work."""
|
||||
from repeater.data_acquisition.sqlite_handler import SQLiteHandler
|
||||
|
||||
handler = SQLiteHandler(tmp_path)
|
||||
|
||||
assert handler.get_daemon_state(STATE_KEY) is None # absent, not an error
|
||||
|
||||
assert handler.set_daemon_state(STATE_KEY, {"last_success_at": 1785372000.0}) is True
|
||||
assert handler.get_daemon_state(STATE_KEY) == {"last_success_at": 1785372000.0}
|
||||
|
||||
# Upsert, not a second row.
|
||||
assert handler.set_daemon_state(STATE_KEY, {"last_success_at": 1785458400.0}) is True
|
||||
assert handler.get_daemon_state(STATE_KEY)["last_success_at"] == 1785458400.0
|
||||
|
||||
# A fresh handler on the same file sees it -- this is the restart path.
|
||||
assert SQLiteHandler(tmp_path).get_daemon_state(STATE_KEY)["last_success_at"] == 1785458400.0
|
||||
|
||||
|
||||
def test_daemon_state_survives_a_corrupt_row(tmp_path):
|
||||
"""Unparseable state must read as "no history", never break startup."""
|
||||
import sqlite3
|
||||
|
||||
from repeater.data_acquisition.sqlite_handler import SQLiteHandler
|
||||
|
||||
handler = SQLiteHandler(tmp_path)
|
||||
handler.set_daemon_state(STATE_KEY, {"last_success_at": 1.0})
|
||||
with sqlite3.connect(handler.sqlite_path) as conn:
|
||||
conn.execute("UPDATE daemon_state SET value_json = ?", ("{not json",))
|
||||
conn.commit()
|
||||
|
||||
assert handler.get_daemon_state(STATE_KEY) is None
|
||||
|
||||
publisher = _publisher({"mqtt_brokers": {}}, handler=_enabled_handler(), storage=handler)
|
||||
publisher._restore_schedule()
|
||||
assert publisher.status()["secs_until_next"] == pytest.approx(STARTUP_GRACE_SECONDS, abs=2)
|
||||
|
||||
|
||||
def test_migration_is_idempotent_on_an_existing_database(tmp_path):
|
||||
"""Migration 14 runs against nibbler's populated DB, not a fresh file."""
|
||||
from repeater.data_acquisition.sqlite_handler import SQLiteHandler
|
||||
|
||||
first = SQLiteHandler(tmp_path)
|
||||
first.set_daemon_state(STATE_KEY, {"last_success_at": 42.0})
|
||||
|
||||
# Re-running migrations must not drop the table or its contents.
|
||||
for _ in range(3):
|
||||
SQLiteHandler(tmp_path)._run_migrations()
|
||||
|
||||
assert SQLiteHandler(tmp_path).get_daemon_state(STATE_KEY) == {"last_success_at": 42.0}
|
||||
|
||||
Reference in New Issue
Block a user