"""node_state.py — Shared mutable node runtime state and node helper functions. Extracted from main.py so that background worker functions and route handlerscan reference the same state objects without importing the full application. _NODE_SYNC_STATE is a reassignable value (SyncWorkerState is replaced whole,not mutated), so callers must use get_sync_state() / set_sync_state() insteadof binding to the name at import time. All other _NODE_* objects are mutable containers (Lock, Event, dict) whoseidentity never changes; importing them directly by name is safe.""" import threadingimport timefrom typing import Any from services.mesh.mesh_infonet_sync_support import SyncWorkerState # ---------------------------------------------------------------------------# Runtime state objects# --------------------------------------------------------------------------- _NODE_RUNTIME_LOCK = threading.RLock()_NODE_SYNC_STOP = threading.Event()_NODE_SYNC_STATE = SyncWorkerState()_NODE_BOOTSTRAP_STATE: dict[str, Any] = { "node_mode": "participant", "manifest_loaded": False, "manifest_signer_id": "", "manifest_valid_until": 0, "bootstrap_peer_count": 0, "sync_peer_count": 0, "push_peer_count": 0, "operator_peer_count": 0, "last_bootstrap_error": "",}_NODE_PUSH_STATE: dict[str, Any] = { "last_event_id": "", "last_push_ok_at": 0, "last_push_error": "", "last_results": [],} # ---------------------------------------------------------------------------# Getter / setter for _NODE_SYNC_STATE## Use these instead of globals()["_NODE_SYNC_STATE"] = ... in any module that# imports this package. The setter modifies *this* module's namespace so# subsequent get_sync_state() calls see the new value regardless of which# module calls set_sync_state().# --------------------------------------------------------------------------- def get_sync_state() -> SyncWorkerState: return _NODE_SYNC_STATE def set_sync_state(state: SyncWorkerState) -> None: global _NODE_SYNC_STATE _NODE_SYNC_STATE = state # ---------------------------------------------------------------------------# Node helper functions## These were in main.py but are needed by both route handlers and background# workers, so they live here to avoid circular imports.# --------------------------------------------------------------------------- def _current_node_mode() -> str: from services.config import get_settings mode = str(get_settings().MESH_NODE_MODE or "participant").strip().lower() if mode not in {"participant", "relay", "perimeter"}: return "participant" return mode def _node_runtime_supported() -> bool: return _current_node_mode() in {"participant", "relay"} def _node_activation_enabled() -> bool: from services.node_settings import read_node_settings try: settings = read_node_settings() except Exception: return False return bool(settings.get("enabled", False)) def _participant_node_enabled() -> bool: return _node_runtime_supported() and _node_activation_enabled() def _node_runtime_snapshot() -> dict[str, Any]: with _NODE_RUNTIME_LOCK: return { "node_mode": _current_node_mode(), "node_enabled": _participant_node_enabled(), "private_transport_required": _infonet_private_transport_required(), "bootstrap": {**dict(_NODE_BOOTSTRAP_STATE), "node_mode": _current_node_mode()}, "sync_runtime": get_sync_state().to_dict(), "push_runtime": dict(_NODE_PUSH_STATE), } def _set_node_sync_disabled_state(*, current_head: str = "") -> SyncWorkerState: return SyncWorkerState( current_head=str(current_head or ""), last_outcome="disabled", ) def _set_participant_node_enabled(enabled: bool) -> dict[str, Any]: from services.mesh.mesh_hashchain import infonet from services.node_settings import write_node_settings settings = write_node_settings(enabled=bool(enabled)) current_head = str(infonet.head_hash or "") with _NODE_RUNTIME_LOCK: _NODE_BOOTSTRAP_STATE["node_mode"] = _current_node_mode() set_sync_state( SyncWorkerState(current_head=current_head) if bool(enabled) and _node_runtime_supported() else _set_node_sync_disabled_state(current_head=current_head) ) return { **settings, "node_mode": _current_node_mode(), "node_enabled": _participant_node_enabled(), } def _infonet_private_transport_required() -> bool: from services.config import get_settings return not bool(getattr(get_settings(), "MESH_INFONET_ALLOW_CLEARNET_SYNC", False)) def _infonet_private_transport_error() -> str: return "private Infonet requires onion/RNS transport; no clearnet sync fallback" def _is_private_infonet_transport(transport: str) -> bool: return str(transport or "").strip().lower() in {"onion", "rns"} def _configured_bootstrap_seed_peer_urls() -> list[str]: from services.config import get_settings from services.mesh.mesh_router import parse_configured_relay_peers settings = get_settings() primary = str(getattr(settings, "MESH_BOOTSTRAP_SEED_PEERS", "") or "").strip() legacy = str(getattr(settings, "MESH_DEFAULT_SYNC_PEERS", "") or "").strip() return parse_configured_relay_peers(primary or legacy) def _refresh_node_peer_store(*, now: float | None = None) -> dict[str, Any]: from services.config import get_settings from services.mesh.mesh_bootstrap_manifest import load_bootstrap_manifest_from_settings from services.mesh.mesh_peer_store import ( DEFAULT_PEER_STORE_PATH, PeerStore, make_bootstrap_peer_record, make_push_peer_record, make_sync_peer_record, ) from services.mesh.mesh_router import ( configured_relay_peer_urls, parse_configured_relay_peers, peer_transport_kind, ) timestamp = int(now if now is not None else time.time()) mode = _current_node_mode() store = PeerStore(DEFAULT_PEER_STORE_PATH) try: store.load() except Exception: store = PeerStore(DEFAULT_PEER_STORE_PATH) private_transport_required = _infonet_private_transport_required() operator_peers = configured_relay_peer_urls() bootstrap_seed_peers = _configured_bootstrap_seed_peer_urls() skipped_clearnet_peers = 0 for peer_url in operator_peers: transport = peer_transport_kind(peer_url) if not transport: continue if private_transport_required and not _is_private_infonet_transport(transport): skipped_clearnet_peers += 1 continue store.upsert( make_sync_peer_record( peer_url=peer_url, transport=transport, role="relay", source="operator", now=timestamp, ) ) store.upsert( make_push_peer_record( peer_url=peer_url, transport=transport, role="relay", source="operator", now=timestamp, ) ) operator_peer_set = set(operator_peers) for peer_url in bootstrap_seed_peers: if peer_url in operator_peer_set: continue transport = peer_transport_kind(peer_url) if not transport: continue if private_transport_required and not _is_private_infonet_transport(transport): skipped_clearnet_peers += 1 continue store.upsert( make_bootstrap_peer_record( peer_url=peer_url, transport=transport, role="seed", label="ShadowBroker bootstrap seed", signer_id="shadowbroker-bootstrap", now=timestamp, ) ) store.upsert( make_sync_peer_record( peer_url=peer_url, transport=transport, role="seed", source="bundle", label="ShadowBroker bootstrap seed", signer_id="shadowbroker-bootstrap", now=timestamp, ) ) manifest = None bootstrap_error = "" try: manifest = load_bootstrap_manifest_from_settings(now=timestamp) except Exception as exc: bootstrap_error = str(exc or "").strip() if manifest is not None: for peer in manifest.peers: if private_transport_required and not _is_private_infonet_transport(peer.transport): skipped_clearnet_peers += 1 continue store.upsert( make_bootstrap_peer_record( peer_url=peer.peer_url, transport=peer.transport, role=peer.role, label=peer.label, signer_id=manifest.signer_id, now=timestamp, ) ) store.upsert( make_sync_peer_record( peer_url=peer.peer_url, transport=peer.transport, role=peer.role, source="bootstrap_promoted", label=peer.label, signer_id=manifest.signer_id, now=timestamp, ) ) if private_transport_required and skipped_clearnet_peers and not bootstrap_error: bootstrap_error = _infonet_private_transport_error() store.save() bootstrap_records = store.records_for_bucket("bootstrap") sync_records = store.records_for_bucket("sync") push_records = store.records_for_bucket("push") if private_transport_required: bootstrap_records = [record for record in bootstrap_records if _is_private_infonet_transport(record.transport)] sync_records = [record for record in sync_records if _is_private_infonet_transport(record.transport)] push_records = [record for record in push_records if _is_private_infonet_transport(record.transport)] snapshot = { "node_mode": mode, "private_transport_required": private_transport_required, "skipped_clearnet_peer_count": skipped_clearnet_peers, "manifest_loaded": manifest is not None, "manifest_signer_id": manifest.signer_id if manifest is not None else "", "manifest_valid_until": int(manifest.valid_until or 0) if manifest is not None else 0, "bootstrap_peer_count": len(bootstrap_records), "sync_peer_count": len(sync_records), "push_peer_count": len(push_records), "operator_peer_count": len(operator_peers), "bootstrap_seed_peer_count": len(bootstrap_seed_peers), "default_sync_peer_count": len(bootstrap_seed_peers), "last_bootstrap_error": bootstrap_error, } with _NODE_RUNTIME_LOCK: _NODE_BOOTSTRAP_STATE.update(snapshot) return snapshot def _materialize_local_infonet_state() -> None: from services.mesh.mesh_hashchain import infonet infonet.ensure_materialized()
"""node_state.py — Shared mutable node runtime state and node helper functions.
Extracted from main.py so that background worker functions and route handlers
can reference the same state objects without importing the full application.
_NODE_SYNC_STATE is a reassignable value (SyncWorkerState is replaced whole,
not mutated), so callers must use get_sync_state() / set_sync_state() instead
of binding to the name at import time.
All other _NODE_* objects are mutable containers (Lock, Event, dict) whose
identity never changes; importing them directly by name is safe.
"""
import threading
import time
from typing import Any
from services.mesh.mesh_infonet_sync_support import SyncWorkerState
# ---------------------------------------------------------------------------
# Runtime state objects
# ---------------------------------------------------------------------------
_NODE_RUNTIME_LOCK = threading.RLock()
_NODE_SYNC_STOP = threading.Event()
_NODE_SYNC_STATE = SyncWorkerState()
_NODE_BOOTSTRAP_STATE: dict[str, Any] = {
"node_mode": "participant",
"manifest_loaded": False,
"manifest_signer_id": "",
"manifest_valid_until": 0,
"bootstrap_peer_count": 0,
"sync_peer_count": 0,
"push_peer_count": 0,
"operator_peer_count": 0,
"last_bootstrap_error": "",
}
_NODE_PUSH_STATE: dict[str, Any] = {
"last_event_id": "",
"last_push_ok_at": 0,
"last_push_error": "",
"last_results": [],
}
# ---------------------------------------------------------------------------
# Getter / setter for _NODE_SYNC_STATE
#
# Use these instead of globals()["_NODE_SYNC_STATE"] = ... in any module that
# imports this package. The setter modifies *this* module's namespace so
# subsequent get_sync_state() calls see the new value regardless of which
# module calls set_sync_state().
# ---------------------------------------------------------------------------
def get_sync_state() -> SyncWorkerState:
return _NODE_SYNC_STATE
def set_sync_state(state: SyncWorkerState) -> None:
global _NODE_SYNC_STATE
_NODE_SYNC_STATE = state
# ---------------------------------------------------------------------------
# Node helper functions
#
# These were in main.py but are needed by both route handlers and background
# workers, so they live here to avoid circular imports.
# ---------------------------------------------------------------------------
def _current_node_mode() -> str:
from services.config import get_settings
mode = str(get_settings().MESH_NODE_MODE or "participant").strip().lower()
if mode not in {"participant", "relay", "perimeter"}:
return "participant"
return mode
def _node_runtime_supported() -> bool:
return _current_node_mode() in {"participant", "relay"}
def _node_activation_enabled() -> bool:
from services.node_settings import read_node_settings
try:
settings = read_node_settings()
except Exception:
return False
return bool(settings.get("enabled", False))
def _participant_node_enabled() -> bool:
return _node_runtime_supported() and _node_activation_enabled()
def _node_runtime_snapshot() -> dict[str, Any]:
with _NODE_RUNTIME_LOCK:
return {
"node_mode": _current_node_mode(),
"node_enabled": _participant_node_enabled(),
"private_transport_required": _infonet_private_transport_required(),
"bootstrap": {**dict(_NODE_BOOTSTRAP_STATE), "node_mode": _current_node_mode()},
"sync_runtime": get_sync_state().to_dict(),
"push_runtime": dict(_NODE_PUSH_STATE),
}
def _set_node_sync_disabled_state(*, current_head: str = "") -> SyncWorkerState:
return SyncWorkerState(
current_head=str(current_head or ""),
last_outcome="disabled",
)
def _set_participant_node_enabled(enabled: bool) -> dict[str, Any]:
from services.mesh.mesh_hashchain import infonet
from services.node_settings import write_node_settings
settings = write_node_settings(enabled=bool(enabled))
current_head = str(infonet.head_hash or "")
with _NODE_RUNTIME_LOCK:
_NODE_BOOTSTRAP_STATE["node_mode"] = _current_node_mode()
set_sync_state(
SyncWorkerState(current_head=current_head)
if bool(enabled) and _node_runtime_supported()
else _set_node_sync_disabled_state(current_head=current_head)
)
return {
**settings,
"node_mode": _current_node_mode(),
"node_enabled": _participant_node_enabled(),
}
def _infonet_private_transport_required() -> bool:
from services.config import get_settings
return not bool(getattr(get_settings(), "MESH_INFONET_ALLOW_CLEARNET_SYNC", False))
def _infonet_private_transport_error() -> str:
return "private Infonet requires onion/RNS transport; no clearnet sync fallback"
def _is_private_infonet_transport(transport: str) -> bool:
return str(transport or "").strip().lower() in {"onion", "rns"}
def _configured_bootstrap_seed_peer_urls() -> list[str]:
from services.config import get_settings
from services.mesh.mesh_router import parse_configured_relay_peers
settings = get_settings()
primary = str(getattr(settings, "MESH_BOOTSTRAP_SEED_PEERS", "") or "").strip()
legacy = str(getattr(settings, "MESH_DEFAULT_SYNC_PEERS", "") or "").strip()
return parse_configured_relay_peers(primary or legacy)
def _refresh_node_peer_store(*, now: float | None = None) -> dict[str, Any]:
from services.config import get_settings
from services.mesh.mesh_bootstrap_manifest import load_bootstrap_manifest_from_settings
from services.mesh.mesh_peer_store import (
DEFAULT_PEER_STORE_PATH,
PeerStore,
make_bootstrap_peer_record,
make_push_peer_record,
make_sync_peer_record,
)
from services.mesh.mesh_router import (
configured_relay_peer_urls,
parse_configured_relay_peers,
peer_transport_kind,
)
timestamp = int(now if now is not None else time.time())
mode = _current_node_mode()
store = PeerStore(DEFAULT_PEER_STORE_PATH)
try:
store.load()
except Exception:
store = PeerStore(DEFAULT_PEER_STORE_PATH)
private_transport_required = _infonet_private_transport_required()
operator_peers = configured_relay_peer_urls()
bootstrap_seed_peers = _configured_bootstrap_seed_peer_urls()
skipped_clearnet_peers = 0
for peer_url in operator_peers:
transport = peer_transport_kind(peer_url)
if not transport:
continue
if private_transport_required and not _is_private_infonet_transport(transport):
skipped_clearnet_peers += 1
continue
store.upsert(
make_sync_peer_record(
peer_url=peer_url,
transport=transport,
role="relay",
source="operator",
now=timestamp,
)
)
store.upsert(
make_push_peer_record(
peer_url=peer_url,
transport=transport,
role="relay",
source="operator",
now=timestamp,
)
)
operator_peer_set = set(operator_peers)
for peer_url in bootstrap_seed_peers:
if peer_url in operator_peer_set:
continue
transport = peer_transport_kind(peer_url)
if not transport:
continue
if private_transport_required and not _is_private_infonet_transport(transport):
skipped_clearnet_peers += 1
continue
store.upsert(
make_bootstrap_peer_record(
peer_url=peer_url,
transport=transport,
role="seed",
label="ShadowBroker bootstrap seed",
signer_id="shadowbroker-bootstrap",
now=timestamp,
)
)
store.upsert(
make_sync_peer_record(
peer_url=peer_url,
transport=transport,
role="seed",
source="bundle",
label="ShadowBroker bootstrap seed",
signer_id="shadowbroker-bootstrap",
now=timestamp,
)
)
manifest = None
bootstrap_error = ""
try:
manifest = load_bootstrap_manifest_from_settings(now=timestamp)
except Exception as exc:
bootstrap_error = str(exc or "").strip()
if manifest is not None:
for peer in manifest.peers:
if private_transport_required and not _is_private_infonet_transport(peer.transport):
skipped_clearnet_peers += 1
continue
store.upsert(
make_bootstrap_peer_record(
peer_url=peer.peer_url,
transport=peer.transport,
role=peer.role,
label=peer.label,
signer_id=manifest.signer_id,
now=timestamp,
)
)
store.upsert(
make_sync_peer_record(
peer_url=peer.peer_url,
transport=peer.transport,
role=peer.role,
source="bootstrap_promoted",
label=peer.label,
signer_id=manifest.signer_id,
now=timestamp,
)
)
if private_transport_required and skipped_clearnet_peers and not bootstrap_error:
bootstrap_error = _infonet_private_transport_error()
store.save()
bootstrap_records = store.records_for_bucket("bootstrap")
sync_records = store.records_for_bucket("sync")
push_records = store.records_for_bucket("push")
if private_transport_required:
bootstrap_records = [record for record in bootstrap_records if _is_private_infonet_transport(record.transport)]
sync_records = [record for record in sync_records if _is_private_infonet_transport(record.transport)]
push_records = [record for record in push_records if _is_private_infonet_transport(record.transport)]
snapshot = {
"node_mode": mode,
"private_transport_required": private_transport_required,
"skipped_clearnet_peer_count": skipped_clearnet_peers,
"manifest_loaded": manifest is not None,
"manifest_signer_id": manifest.signer_id if manifest is not None else "",
"manifest_valid_until": int(manifest.valid_until or 0) if manifest is not None else 0,
"bootstrap_peer_count": len(bootstrap_records),
"sync_peer_count": len(sync_records),
"push_peer_count": len(push_records),
"operator_peer_count": len(operator_peers),
"bootstrap_seed_peer_count": len(bootstrap_seed_peers),
"default_sync_peer_count": len(bootstrap_seed_peers),
"last_bootstrap_error": bootstrap_error,
}
with _NODE_RUNTIME_LOCK:
_NODE_BOOTSTRAP_STATE.update(snapshot)
return snapshot
def _materialize_local_infonet_state() -> None:
from services.mesh.mesh_hashchain import infonet
infonet.ensure_materialized()