feat(propagation): implement local propagation node management with start, stop, and restart functionality; add API endpoints for controlling propagation nodes and retrieving their stats

This commit is contained in:
Ivan committed 2026-04-15 03:11:03 -05:00
1 parent fb775b74b3
commit 7fd826abb0
10 files changed
+481 -45

No files matched your search

+213 -4
View File
@@ -1636,14 +1636,13 @@ class ReticulumMeshChat:
return
while self.running and ctx.running and ctx.session_id == session_id:
auto_sync_interval_seconds = (
ctx.config.lxmf_preferred_propagation_node_auto_sync_interval_seconds.get()
)
auto_sync_interval_seconds = ctx.config.lxmf_preferred_propagation_node_auto_sync_interval_seconds.get()
last_synced_at = (
ctx.config.lxmf_preferred_propagation_node_last_synced_at.get()
)
should_sync = interval_action_due(
auto_sync_interval_seconds is not None and auto_sync_interval_seconds > 0,
auto_sync_interval_seconds is not None
and auto_sync_interval_seconds > 0,
last_synced_at,
auto_sync_interval_seconds,
time.time(),
@@ -1951,6 +1950,108 @@ class ReticulumMeshChat:
f"failed to enable or disable propagation node for {ctx.identity_hash}",
)
def stop_local_propagation_node(self, context=None):
ctx = context or self.current_context
if not ctx:
return
self.enable_local_propagation_node(False, context=ctx)
def restart_local_propagation_node(self, context=None):
ctx = context or self.current_context
if not ctx:
return
self.stop_local_propagation_node(context=ctx)
self.enable_local_propagation_node(True, context=ctx)
def get_local_propagation_node_stats(self, context=None):
ctx = context or self.current_context
if not ctx:
return None
router = ctx.message_router
is_running = bool(getattr(router, "propagation_node", False))
stats = None
if is_running:
with contextlib.suppress(Exception):
stats = router.compile_stats()
def _numeric(value, default=0):
return value if isinstance(value, (int, float)) else default
destination_hash_raw = getattr(
ctx.message_router.propagation_destination,
"hexhash",
None,
)
if destination_hash_raw is None:
destination_hash_raw = getattr(
ctx.message_router.propagation_destination,
"hash",
None,
)
if isinstance(destination_hash_raw, bytes):
destination_hash = destination_hash_raw.hex()
elif isinstance(destination_hash_raw, str):
destination_hash = destination_hash_raw
else:
destination_hash = None
message_store = stats.get("messagestore", {}) if isinstance(stats, dict) else {}
clients = stats.get("clients", {}) if isinstance(stats, dict) else {}
uptime = _numeric(stats.get("uptime", 0)) if isinstance(stats, dict) else 0
delivery_limit = (
_numeric(stats.get("delivery_limit", 0))
if isinstance(stats, dict)
else _numeric(getattr(router, "delivery_per_transfer_limit", 0))
)
propagation_limit = (
_numeric(stats.get("propagation_limit", 0))
if isinstance(stats, dict)
else _numeric(getattr(router, "propagation_per_transfer_limit", 0))
)
sync_limit = (
_numeric(stats.get("sync_limit", 0))
if isinstance(stats, dict)
else _numeric(getattr(router, "propagation_per_sync_limit", 0))
)
return {
"is_running": is_running,
"identity_hash": ctx.identity.hash.hex(),
"destination_hash": destination_hash,
"uptime_seconds": int(uptime) if uptime else 0,
"messagestore_count": message_store.get("count", 0),
"messagestore_bytes": message_store.get("bytes", 0),
"messagestore_limit_bytes": message_store.get("limit"),
"client_messages_received": clients.get(
"client_propagation_messages_received",
0,
),
"client_messages_served": clients.get(
"client_propagation_messages_served",
0,
),
"static_peers": stats.get("static_peers", 0)
if isinstance(stats, dict)
else 0,
"discovered_peers": (
stats.get("discovered_peers", 0) if isinstance(stats, dict) else 0
),
"total_peers": stats.get("total_peers", 0)
if isinstance(stats, dict)
else 0,
"max_peers": stats.get("max_peers") if isinstance(stats, dict) else None,
"delivery_limit_bytes": int(delivery_limit * 1000),
"propagation_limit_bytes": int(propagation_limit * 1000),
"sync_limit_bytes": int(sync_limit * 1000),
"target_stamp_cost": _numeric(
(
stats.get("target_stamp_cost", 0)
if isinstance(stats, dict)
else getattr(router, "propagation_stamp_cost", 0)
),
),
}
def _get_reticulum_section(self):
try:
if hasattr(self, "reticulum") and self.reticulum:
@@ -6747,6 +6848,7 @@ class ReticulumMeshChat:
],
"messages_hidden": sync_metrics["messages_hidden"],
},
"local_propagation_node": self.get_local_propagation_node_stats(),
},
)
@@ -6782,9 +6884,34 @@ class ReticulumMeshChat:
},
)
@routes.post("/api/v1/lxmf/propagation-node/stop")
async def propagation_node_stop(request):
self.config.lxmf_local_propagation_node_enabled.set(False)
self.stop_local_propagation_node()
AsyncUtils.run_async(self.send_config_to_websocket_clients())
return web.json_response(
{
"message": "Local propagation node stopped",
"local_propagation_node": self.get_local_propagation_node_stats(),
},
)
@routes.post("/api/v1/lxmf/propagation-node/restart")
async def propagation_node_restart(request):
self.config.lxmf_local_propagation_node_enabled.set(True)
self.restart_local_propagation_node()
AsyncUtils.run_async(self.send_config_to_websocket_clients())
return web.json_response(
{
"message": "Local propagation node restarted",
"local_propagation_node": self.get_local_propagation_node_stats(),
},
)
# serve propagation nodes
@routes.get("/api/v1/lxmf/propagation-nodes")
async def propagation_nodes_get(request):
ctx = self.current_context
# get query params
limit = request.query.get("limit", None)
@@ -6800,6 +6927,27 @@ class ReticulumMeshChat:
# process announces
lxmf_propagation_nodes = []
local_identity_hash = ctx.identity.hash.hex() if ctx else None
local_destination_hash_raw = (
getattr(ctx.message_router.propagation_destination, "hexhash", None)
if ctx
else None
)
if local_destination_hash_raw is None and ctx:
local_destination_hash_raw = getattr(
ctx.message_router.propagation_destination,
"hash",
None,
)
if isinstance(local_destination_hash_raw, bytes):
local_destination_hash = local_destination_hash_raw.hex()
elif isinstance(local_destination_hash_raw, str):
local_destination_hash = local_destination_hash_raw
else:
local_destination_hash = None
local_stats = (
self.get_local_propagation_node_stats(context=ctx) if ctx else None
)
for announce in results:
# find an lxmf.delivery announce for the same identity hash, so we can use that as an "operater by" name
lxmf_delivery_results = self.database.announces.get_filtered_announces(
@@ -6866,11 +7014,49 @@ class ReticulumMeshChat:
"operator_display_name": operator_display_name,
"is_propagation_enabled": is_propagation_enabled,
"per_transfer_limit": per_transfer_limit,
"is_local_node": (
announce["identity_hash"] == local_identity_hash
or announce["destination_hash"] == local_destination_hash
),
"local_node_stats": (
local_stats
if announce["identity_hash"] == local_identity_hash
or announce["destination_hash"] == local_destination_hash
else None
),
"created_at": created_at,
"updated_at": updated_at,
},
)
if (
ctx is not None
and local_destination_hash is not None
and not any(
node["destination_hash"] == local_destination_hash
for node in lxmf_propagation_nodes
)
):
now_iso = datetime.now(UTC).isoformat()
lxmf_propagation_nodes.insert(
0,
{
"destination_hash": local_destination_hash,
"identity_hash": local_identity_hash,
"operator_display_name": ctx.config.display_name.get(),
"is_propagation_enabled": ctx.config.lxmf_local_propagation_node_enabled.get(),
"per_transfer_limit": int(
getattr(
ctx.message_router, "propagation_per_transfer_limit", 0
),
),
"is_local_node": True,
"local_node_stats": local_stats,
"created_at": now_iso,
"updated_at": now_iso,
},
)
return web.json_response(
{
"lxmf_propagation_nodes": lxmf_propagation_nodes,
@@ -10199,6 +10385,26 @@ class ReticulumMeshChat:
)
self.config.auto_send_failed_messages_to_propagation_node.set(value)
if "lxmf_delivery_transfer_limit_in_bytes" in data:
value = int(data["lxmf_delivery_transfer_limit_in_bytes"])
value = max(1000, min(value, 1000 * 1000 * 100))
self.config.lxmf_delivery_transfer_limit_in_bytes.set(value)
self.message_router.delivery_per_transfer_limit = value / 1000
if "lxmf_propagation_transfer_limit_in_bytes" in data:
value = int(data["lxmf_propagation_transfer_limit_in_bytes"])
value = max(1000, min(value, 1000 * 1000 * 100))
self.config.lxmf_propagation_transfer_limit_in_bytes.set(value)
self.message_router.propagation_per_transfer_limit = value / 1000
if self.config.lxmf_local_propagation_node_enabled.get():
self.message_router.announce_propagation_node()
if "lxmf_propagation_sync_limit_in_bytes" in data:
value = int(data["lxmf_propagation_sync_limit_in_bytes"])
value = max(1000, min(value, 1000 * 1000 * 500))
self.config.lxmf_propagation_sync_limit_in_bytes.set(value)
self.message_router.propagation_per_sync_limit = value / 1000
if "show_suggested_community_interfaces" in data:
value = self._parse_bool(data["show_suggested_community_interfaces"])
self.config.show_suggested_community_interfaces.set(value)
@@ -11623,6 +11829,9 @@ class ReticulumMeshChat:
"allow_auto_resending_failed_messages_with_attachments": ctx.config.allow_auto_resending_failed_messages_with_attachments.get(),
"auto_send_failed_messages_to_propagation_node": ctx.config.auto_send_failed_messages_to_propagation_node.get(),
"show_suggested_community_interfaces": ctx.config.show_suggested_community_interfaces.get(),
"lxmf_delivery_transfer_limit_in_bytes": ctx.config.lxmf_delivery_transfer_limit_in_bytes.get(),
"lxmf_propagation_transfer_limit_in_bytes": ctx.config.lxmf_propagation_transfer_limit_in_bytes.get(),
"lxmf_propagation_sync_limit_in_bytes": ctx.config.lxmf_propagation_sync_limit_in_bytes.get(),
"lxmf_local_propagation_node_enabled": ctx.config.lxmf_local_propagation_node_enabled.get(),
"lxmf_local_propagation_node_address_hash": ctx.message_router.propagation_destination.hexhash,
"lxmf_preferred_propagation_node_destination_hash": ctx.config.lxmf_preferred_propagation_node_destination_hash.get(),
+10
View File
@@ -43,6 +43,16 @@ class ConfigManager:
"lxmf_delivery_transfer_limit_in_bytes",
1000 * 1000 * 10,
) # 10MB
self.lxmf_propagation_transfer_limit_in_bytes = self.IntConfig(
self,
"lxmf_propagation_transfer_limit_in_bytes",
1000 * 256,
) # 256KB (LXMF default)
self.lxmf_propagation_sync_limit_in_bytes = self.IntConfig(
self,
"lxmf_propagation_sync_limit_in_bytes",
1000 * 10240,
) # 10MB (LXMF default)
self.lxmf_preferred_propagation_node_destination_hash = self.StringConfig(
self,
"lxmf_preferred_propagation_node_destination_hash",
@@ -210,6 +210,12 @@ class IdentityContext:
self.message_router.delivery_per_transfer_limit = (
self.config.lxmf_delivery_transfer_limit_in_bytes.get() / 1000
)
self.message_router.propagation_per_transfer_limit = (
self.config.lxmf_propagation_transfer_limit_in_bytes.get() / 1000
)
self.message_router.propagation_per_sync_limit = (
self.config.lxmf_propagation_sync_limit_in_bytes.get() / 1000
)
# Register LXMF delivery identity
inbound_stamp_cost = self.config.lxmf_inbound_stamp_cost.get()
+1
View File
@@ -92,6 +92,7 @@ def convert_propagation_node_state_to_string(state):
LXMRouter.PR_NO_IDENTITY_RCVD: "no_identity_received",
LXMRouter.PR_NO_ACCESS: "no_access",
LXMRouter.PR_FAILED: "failed",
LXMRouter.PR_PATH_TIMEOUT: "path_timeout",
}
if state in state_map:
+2 -2
View File
@@ -733,7 +733,7 @@ export default {
deep: true,
},
},
beforeUnmount() {
beforeUnmount() {
if (typeof this._shellAuthWatchStop === "function") {
this._shellAuthWatchStop();
this._shellAuthWatchStop = null;
@@ -1364,7 +1364,7 @@ export default {
ToastUtils.error(
this.$t("app.sync_error", {
status: this.propagationSyncStatusLabel(status),
}),
})
);
}
};
@@ -570,8 +570,7 @@ export default {
if (this.saveTimeouts.propagationLimit) clearTimeout(this.saveTimeouts.propagationLimit);
this.saveTimeouts.propagationLimit = setTimeout(async () => {
await this.updateConfig({
lxmf_propagation_transfer_limit_in_bytes:
this.config.lxmf_propagation_transfer_limit_in_bytes,
lxmf_propagation_transfer_limit_in_bytes: this.config.lxmf_propagation_transfer_limit_in_bytes,
});
}, 450);
},
@@ -2799,8 +2799,7 @@ export default {
}
this.saveTimeouts.propagation_transfer_limit = setTimeout(async () => {
await this.updateConfig({
lxmf_propagation_transfer_limit_in_bytes:
this.config.lxmf_propagation_transfer_limit_in_bytes,
lxmf_propagation_transfer_limit_in_bytes: this.config.lxmf_propagation_transfer_limit_in_bytes,
});
}, 1000);
},
@@ -380,6 +380,14 @@
"method": "GET",
"path": "/api/v1/lxmf/propagation-node/sync"
},
{
"method": "POST",
"path": "/api/v1/lxmf/propagation-node/stop"
},
{
"method": "POST",
"path": "/api/v1/lxmf/propagation-node/restart"
},
{
"method": "GET",
"path": "/api/v1/lxmf/propagation-nodes"
+156 -35
View File
@@ -6,10 +6,75 @@ vi.mock("../../meshchatx/src/frontend/js/ToastUtils", () => ({
default: {
success: vi.fn(),
error: vi.fn(),
loading: vi.fn(),
dismiss: vi.fn(),
},
}));
describe("App propagation sync metrics", () => {
const syncingStates = [
"path_requested",
"link_establishing",
"link_established",
"request_sent",
"receiving",
"response_received",
];
function makeSyncContext(axiosMock, tOverrides = {}) {
return {
propagationNodeStatus: null,
_propagationSyncPollTimer: null,
propagationSyncLiveToastMessage: App.methods.propagationSyncLiveToastMessage,
propagationSyncStatusLabel: App.methods.propagationSyncStatusLabel,
get isSyncingPropagationNode() {
return syncingStates.includes(this.propagationNodeStatus?.state);
},
async updatePropagationNodeStatus() {
try {
const response = await axiosMock.get("/api/v1/lxmf/propagation-node/status");
this.propagationNodeStatus = response.data.propagation_node_status;
} catch {
// ignore
}
},
async stopSyncingPropagationNode() {},
$t(key, params = {}) {
if (tOverrides[key]) {
return tOverrides[key](params);
}
if (key === "app.sync_complete") {
return `Sync complete. ${params.count} messages received.`;
}
if (key === "app.sync_error") {
return `Sync error: ${params.status}`;
}
if (key === "app.sync_error_generic") {
return "Sync failed";
}
if (key === "app.stop_sync_confirm") {
return "Stop syncing?";
}
if (key === "app.propagation_sync_live") {
return `Syncing: ${params.status} (${params.progress}%)`;
}
if (key.startsWith("app.propagation_sync_state.")) {
const sub = key.slice("app.propagation_sync_state.".length);
const labels = {
path_requested: "Requesting path",
receiving: "Receiving messages",
complete: "Complete",
idle: "Idle",
no_path: "No path to node",
unknown: "Unknown state",
};
return labels[sub] ?? sub;
}
return key;
},
};
}
describe("App propagation sync", () => {
const axiosMock = {
get: vi.fn(),
};
@@ -18,13 +83,14 @@ describe("App propagation sync metrics", () => {
vi.clearAllMocks();
vi.useFakeTimers();
globalThis.api = axiosMock;
window.api = axiosMock;
});
afterEach(() => {
vi.useRealTimers();
});
it("shows detailed sync toast with stored, confirmations and hidden counts", async () => {
it("shows detailed success toast with stored, confirmations and hidden counts", async () => {
axiosMock.get.mockImplementation((url) => {
if (url === "/api/v1/lxmf/propagation-node/sync") {
return Promise.resolve({ data: { message: "Sync is starting" } });
@@ -34,6 +100,7 @@ describe("App propagation sync metrics", () => {
data: {
propagation_node_status: {
state: "complete",
progress: 100,
messages_received: 8,
messages_stored: 3,
delivery_confirmations: 2,
@@ -45,45 +112,99 @@ describe("App propagation sync metrics", () => {
return Promise.resolve({ data: {} });
});
const ctx = {
propagationNodeStatus: null,
get isSyncingPropagationNode() {
return [
"path_requested",
"link_establishing",
"link_established",
"request_sent",
"receiving",
"response_received",
].includes(this.propagationNodeStatus?.state);
},
async updatePropagationNodeStatus() {
return App.methods.updatePropagationNodeStatus.call(this);
},
async stopSyncingPropagationNode() {},
$t(key, params = {}) {
if (key === "app.sync_complete") {
return `Sync complete. ${params.count} messages received.`;
}
if (key === "app.sync_error") {
return `Sync error: ${params.status}`;
}
if (key === "app.sync_error_generic") {
return "Sync failed";
}
if (key === "app.stop_sync_confirm") {
return "Stop syncing?";
}
return key;
},
};
const ctx = makeSyncContext(axiosMock);
await App.methods.syncPropagationNode.call(ctx);
vi.advanceTimersByTime(600);
await vi.runOnlyPendingTimersAsync();
expect(ToastUtils.loading).not.toHaveBeenCalled();
expect(ToastUtils.dismiss).toHaveBeenCalledWith("propagation-sync-status");
expect(ToastUtils.success).toHaveBeenCalledWith(
"Sync complete. 8 messages received. (3 stored, 2 confirmations, 3 hidden)"
);
expect(ToastUtils.error).not.toHaveBeenCalled();
});
it("polls status while syncing and updates live loading toast", async () => {
let statusCalls = 0;
axiosMock.get.mockImplementation((url) => {
if (url === "/api/v1/lxmf/propagation-node/sync") {
return Promise.resolve({ data: { message: "Sync is starting" } });
}
if (url === "/api/v1/lxmf/propagation-node/status") {
statusCalls += 1;
if (statusCalls < 3) {
return Promise.resolve({
data: {
propagation_node_status: {
state: "path_requested",
progress: 12,
messages_received: 0,
messages_stored: 0,
delivery_confirmations: 0,
messages_hidden: 0,
},
},
});
}
return Promise.resolve({
data: {
propagation_node_status: {
state: "complete",
progress: 100,
messages_received: 2,
messages_stored: 1,
delivery_confirmations: 1,
messages_hidden: 0,
},
},
});
}
return Promise.resolve({ data: {} });
});
const ctx = makeSyncContext(axiosMock);
const syncPromise = App.methods.syncPropagationNode.call(ctx);
await vi.runOnlyPendingTimersAsync();
vi.advanceTimersByTime(500);
await vi.runOnlyPendingTimersAsync();
await syncPromise;
expect(statusCalls).toBeGreaterThanOrEqual(3);
expect(ToastUtils.loading).toHaveBeenCalledWith("Syncing: Requesting path (12%)", 0, "propagation-sync-status");
expect(ToastUtils.dismiss).toHaveBeenCalledWith("propagation-sync-status");
expect(ToastUtils.success).toHaveBeenCalled();
});
it("uses translated status in error toast when sync ends in a failure state", async () => {
axiosMock.get.mockImplementation((url) => {
if (url === "/api/v1/lxmf/propagation-node/sync") {
return Promise.resolve({ data: { message: "Sync is starting" } });
}
if (url === "/api/v1/lxmf/propagation-node/status") {
return Promise.resolve({
data: {
propagation_node_status: {
state: "no_path",
progress: 0,
messages_received: 0,
messages_stored: 0,
delivery_confirmations: 0,
messages_hidden: 0,
},
},
});
}
return Promise.resolve({ data: {} });
});
const ctx = makeSyncContext(axiosMock);
await App.methods.syncPropagationNode.call(ctx);
await vi.runOnlyPendingTimersAsync();
expect(ToastUtils.error).toHaveBeenCalledWith("Sync error: No path to node");
expect(ToastUtils.success).not.toHaveBeenCalled();
});
});
@@ -0,0 +1,83 @@
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import PropagationNodesPage from "../../meshchatx/src/frontend/components/propagation-nodes/PropagationNodesPage.vue";
import ToastUtils from "../../meshchatx/src/frontend/js/ToastUtils";
vi.mock("../../meshchatx/src/frontend/js/ToastUtils", () => ({
default: {
success: vi.fn(),
error: vi.fn(),
},
}));
describe("PropagationNodesPage", () => {
const axiosMock = {
post: vi.fn(),
};
beforeEach(() => {
vi.useFakeTimers();
vi.clearAllMocks();
window.api = axiosMock;
});
afterEach(() => {
vi.useRealTimers();
});
it("finds local propagation node from list", () => {
const ctx = {
propagationNodes: [
{ destination_hash: "remote-a", is_local_node: false },
{ destination_hash: "local-node", is_local_node: true },
],
};
const local = PropagationNodesPage.computed.localPropagationNode.call(ctx);
expect(local.destination_hash).toBe("local-node");
});
it("uses local propagation node as preferred", async () => {
const ctx = {
localPropagationNode: { destination_hash: "local-node" },
usePropagationNode: vi.fn(),
};
await PropagationNodesPage.methods.useLocalPropagationNode.call(ctx);
expect(ctx.usePropagationNode).toHaveBeenCalledWith("local-node");
});
it("debounces propagation transfer limit save", async () => {
const ctx = {
config: {
lxmf_propagation_transfer_limit_in_bytes: 123456,
},
saveTimeouts: {
propagationLimit: null,
},
updateConfig: vi.fn().mockResolvedValue(undefined),
};
await PropagationNodesPage.methods.onPropagationTransferLimitChange.call(ctx);
expect(ctx.updateConfig).not.toHaveBeenCalled();
await vi.advanceTimersByTimeAsync(500);
expect(ctx.updateConfig).toHaveBeenCalledWith({
lxmf_propagation_transfer_limit_in_bytes: 123456,
});
});
it("stops and restarts local node via API", async () => {
axiosMock.post.mockResolvedValue({ data: {} });
const ctx = {
getConfig: vi.fn().mockResolvedValue(undefined),
loadPropagationNodes: vi.fn().mockResolvedValue(undefined),
$t: (k) => k,
};
await PropagationNodesPage.methods.stopLocalPropagationNode.call(ctx);
await PropagationNodesPage.methods.restartLocalPropagationNode.call(ctx);
expect(axiosMock.post).toHaveBeenCalledWith("/api/v1/lxmf/propagation-node/stop");
expect(axiosMock.post).toHaveBeenCalledWith("/api/v1/lxmf/propagation-node/restart");
expect(ToastUtils.success).toHaveBeenCalledTimes(2);
});
});