mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-08-28 23:00:41 +00:00
feat(mesh): enhance mesh graph performance and caching
- Implemented caching for multi-byte mesh aggregation, allowing concurrent requests to share computation and reducing CPU load. - Updated the web viewer to coalesce live edge events and refresh the mesh graph every 30 seconds, improving responsiveness on busy networks. - Adjusted logging behavior in the web viewer to respect the configured log level, ensuring efficient logging without excessive duplicate entries. - Added configuration options for mesh graph caching duration, enhancing user control over performance settings.
This commit is contained in:
agessaman
@@ -117,6 +117,12 @@ semantic versioning.
|
||||
lock on SD-card installations.
|
||||
- Linux service installers now use 1GB memory and 200% CPU ceilings, providing
|
||||
Raspberry Pi graph and web-viewer workloads with practical headroom.
|
||||
- The mesh map now coalesces live edge events, serializes graph reloads, pauses
|
||||
hidden-tab refreshes, and caches concurrent multi-byte aggregation requests,
|
||||
preventing an open graph page from creating a CPU and SQLite I/O request
|
||||
storm on busy meshes.
|
||||
- The web-viewer rotating-file handler now honors `[Logging] log_level`, while
|
||||
its journal handler retains an INFO floor to avoid duplicate debug writes.
|
||||
|
||||
### Removed
|
||||
|
||||
|
||||
@@ -1614,6 +1614,11 @@ auto_start = false
|
||||
# true: expose page/nav/API endpoint
|
||||
multibyte_monitor_enabled = false
|
||||
|
||||
# Cache the expensive lifetime multi-byte mesh aggregation for this many seconds.
|
||||
# Concurrent graph requests share one computation. The mesh page also coalesces
|
||||
# live edge events into one refresh every 30 seconds.
|
||||
mesh_graph_cache_seconds = 30
|
||||
|
||||
# Dashboard snapshot refresher
|
||||
# A background thread in the viewer process recomputes the landing page's
|
||||
# statistics on an interval and stores them in two tables (daily_rollup and
|
||||
|
||||
@@ -296,6 +296,12 @@ The installed systemd service allows up to 1GB of memory and 200% CPU (two
|
||||
cores). These are upper limits rather than reserved resources. A USB SSD is
|
||||
still the most effective way to reduce SD-card wear on high-volume nodes.
|
||||
|
||||
The web viewer caches the lifetime multi-byte edge aggregate for 30 seconds and
|
||||
coalesces live mesh events into one refresh per 30 seconds. On unusually large
|
||||
meshes, increase `[Web_Viewer] mesh_graph_cache_seconds` to 60. Keep
|
||||
`[Logging] log_level = INFO` during normal operation; `DEBUG` emits multiple
|
||||
records per observed edge and can create substantial SD-card write traffic.
|
||||
|
||||
## Preset Configurations
|
||||
|
||||
### `balanced` (Default)
|
||||
|
||||
@@ -201,6 +201,7 @@ SECTIONS: dict[str, SectionMeta] = {
|
||||
"sqlite_busy_timeout_ms": KeyMeta(type="int", default="60000"),
|
||||
"sqlite_foreign_keys": KeyMeta(type="bool", default="true"),
|
||||
"sqlite_journal_mode": KeyMeta(default="WAL"),
|
||||
"mesh_graph_cache_seconds": KeyMeta(type="int", default="30"),
|
||||
# Dashboard snapshot refresher — moves the landing page's aggregate
|
||||
# queries off the request path and accumulates trends that outlive the
|
||||
# raw tables' retention.
|
||||
|
||||
+153
-9
@@ -307,6 +307,28 @@ class BotDataViewer:
|
||||
self._contacts_badge_cache_lock = threading.Lock()
|
||||
self._contacts_badge_cache: dict[int | None, tuple[tuple, set[str]]] = {}
|
||||
|
||||
# The multi-byte mesh endpoint derives lifetime edge identity from every
|
||||
# retained multi-byte observed path. Cache that expensive aggregate and
|
||||
# ensure concurrent requests share one computation. View-specific day
|
||||
# and observation filters remain cheap and are applied to the cached
|
||||
# lifetime result.
|
||||
try:
|
||||
mesh_cache_seconds = self.config.getint(
|
||||
'Web_Viewer',
|
||||
'mesh_graph_cache_seconds',
|
||||
fallback=30,
|
||||
)
|
||||
except (configparser.Error, ValueError, TypeError):
|
||||
mesh_cache_seconds = 30
|
||||
self._mesh_graph_cache_seconds = max(5, min(mesh_cache_seconds, 300))
|
||||
self._multibyte_graph_cache_condition = threading.Condition()
|
||||
self._multibyte_graph_cache_edges: list[dict[str, Any]] | None = None
|
||||
self._multibyte_graph_cache_created_at = 0.0
|
||||
self._multibyte_graph_cache_computing = False
|
||||
self._multibyte_graph_cache_failure_at = 0.0
|
||||
self._multibyte_graph_cache_failure: tuple[str, str] | None = None
|
||||
self._multibyte_graph_cache_retry_seconds = 5.0
|
||||
|
||||
# Use [Bot] db_path when [Web_Viewer] db_path is unset
|
||||
bot_db = self.config.get('Bot', 'db_path', fallback='meshcore_bot.db')
|
||||
if (self.config.has_section('Web_Viewer') and self.config.has_option('Web_Viewer', 'db_path')
|
||||
@@ -396,9 +418,20 @@ class BotDataViewer:
|
||||
except (configparser.Error, ValueError, TypeError):
|
||||
pass
|
||||
|
||||
log_level_name = 'INFO'
|
||||
if getattr(self, 'config', None) is not None and self.config.has_section('Logging'):
|
||||
log_level_name = self.config.get(
|
||||
'Logging',
|
||||
'log_level',
|
||||
fallback='INFO',
|
||||
).strip().upper()
|
||||
log_level = getattr(logging, log_level_name, logging.INFO)
|
||||
if not isinstance(log_level, int):
|
||||
log_level = logging.INFO
|
||||
|
||||
# Get or create logger (don't use basicConfig as it may conflict with existing logging)
|
||||
self.logger = logging.getLogger('modern_web_viewer')
|
||||
self.logger.setLevel(logging.DEBUG)
|
||||
self.logger.setLevel(log_level)
|
||||
|
||||
# Remove existing handlers to avoid duplicates
|
||||
self.logger.handlers.clear()
|
||||
@@ -407,7 +440,9 @@ class BotDataViewer:
|
||||
|
||||
# Console handler (captured by journald under systemd)
|
||||
console_handler = logging.StreamHandler()
|
||||
console_handler.setLevel(logging.INFO)
|
||||
# Keep DEBUG out of journald even when explicitly enabled for the
|
||||
# rotating diagnostic file, avoiding duplicate high-volume SD writes.
|
||||
console_handler.setLevel(max(log_level, logging.INFO))
|
||||
console_handler.setFormatter(formatter)
|
||||
self.logger.addHandler(console_handler)
|
||||
|
||||
@@ -429,7 +464,7 @@ class BotDataViewer:
|
||||
backupCount=log_backup_count,
|
||||
encoding='utf-8',
|
||||
)
|
||||
file_handler.setLevel(logging.DEBUG)
|
||||
file_handler.setLevel(log_level)
|
||||
file_handler.setFormatter(formatter)
|
||||
self.logger.addHandler(file_handler)
|
||||
self.logger.info(
|
||||
@@ -609,10 +644,16 @@ class BotDataViewer:
|
||||
conn.close()
|
||||
|
||||
def _derive_multibyte_evidence_edges(
|
||||
self, days: int | None = None, min_observations: int | None = None
|
||||
self,
|
||||
days: int | None = None,
|
||||
min_observations: int | None = None,
|
||||
*,
|
||||
force_refresh: bool = False,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Return lifetime-derived multi-byte edges filtered for the API view."""
|
||||
all_edges = self._aggregate_multibyte_evidence_edges()
|
||||
all_edges = self._aggregate_multibyte_evidence_edges(
|
||||
force_refresh=force_refresh
|
||||
)
|
||||
return self._filter_multibyte_evidence_edges(
|
||||
all_edges,
|
||||
days=days,
|
||||
@@ -620,10 +661,16 @@ class BotDataViewer:
|
||||
)
|
||||
|
||||
def _derive_multibyte_evidence_graph(
|
||||
self, days: int | None = None, min_observations: int | None = None
|
||||
self,
|
||||
days: int | None = None,
|
||||
min_observations: int | None = None,
|
||||
*,
|
||||
force_refresh: bool = False,
|
||||
) -> tuple[list[dict[str, Any]], int]:
|
||||
"""Return filtered edges plus the lifetime graph's prefix resolution."""
|
||||
all_edges = self._aggregate_multibyte_evidence_edges()
|
||||
all_edges = self._aggregate_multibyte_evidence_edges(
|
||||
force_refresh=force_refresh
|
||||
)
|
||||
prefix_hex_chars = max(
|
||||
(len(edge['from_prefix']) for edge in all_edges),
|
||||
default=2,
|
||||
@@ -675,7 +722,101 @@ class BotDataViewer:
|
||||
result.append(edge)
|
||||
return result
|
||||
|
||||
def _aggregate_multibyte_evidence_edges(self) -> list[dict[str, Any]]:
|
||||
def _aggregate_multibyte_evidence_edges(
|
||||
self, *, force_refresh: bool = False
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Return a bounded-age, single-flight lifetime multi-byte aggregate."""
|
||||
def previous_failure(message: str) -> RuntimeError:
|
||||
failure = self._multibyte_graph_cache_failure
|
||||
if failure is None:
|
||||
return RuntimeError(message)
|
||||
failure_type, failure_message = failure
|
||||
return RuntimeError(
|
||||
f"{message}: {failure_type}: {failure_message}"
|
||||
)
|
||||
|
||||
now = time.monotonic()
|
||||
with self._multibyte_graph_cache_condition:
|
||||
cached = self._multibyte_graph_cache_edges
|
||||
cache_age = now - self._multibyte_graph_cache_created_at
|
||||
failure_age = now - self._multibyte_graph_cache_failure_at
|
||||
retry_suppressed = (
|
||||
self._multibyte_graph_cache_failure is not None
|
||||
and failure_age < self._multibyte_graph_cache_retry_seconds
|
||||
)
|
||||
if retry_suppressed:
|
||||
if cached is not None and not force_refresh:
|
||||
return cached
|
||||
raise previous_failure(
|
||||
"Multi-byte mesh aggregation retry suppressed after failure"
|
||||
)
|
||||
if (
|
||||
not force_refresh
|
||||
and cached is not None
|
||||
and cache_age < self._mesh_graph_cache_seconds
|
||||
):
|
||||
return cached
|
||||
|
||||
if self._multibyte_graph_cache_computing:
|
||||
# Prefer a slightly stale result to making concurrent clients
|
||||
# duplicate the same expensive SQLite aggregation.
|
||||
if cached is not None and not force_refresh:
|
||||
return cached
|
||||
while self._multibyte_graph_cache_computing:
|
||||
self._multibyte_graph_cache_condition.wait()
|
||||
cached = self._multibyte_graph_cache_edges
|
||||
if self._multibyte_graph_cache_failure is not None:
|
||||
if cached is not None and not force_refresh:
|
||||
return cached
|
||||
raise previous_failure(
|
||||
"Concurrent multi-byte mesh aggregation failed"
|
||||
)
|
||||
if cached is not None:
|
||||
return cached
|
||||
|
||||
self._multibyte_graph_cache_computing = True
|
||||
stale = cached
|
||||
|
||||
started_at = time.monotonic()
|
||||
try:
|
||||
computed = self._compute_multibyte_evidence_edges()
|
||||
except Exception as exc:
|
||||
self.logger.warning(
|
||||
"Multi-byte mesh aggregation failed%s",
|
||||
"; serving cached data"
|
||||
if stale is not None and not force_refresh
|
||||
else "",
|
||||
exc_info=True,
|
||||
)
|
||||
with self._multibyte_graph_cache_condition:
|
||||
self._multibyte_graph_cache_failure = (
|
||||
type(exc).__name__,
|
||||
str(exc),
|
||||
)
|
||||
self._multibyte_graph_cache_failure_at = time.monotonic()
|
||||
self._multibyte_graph_cache_computing = False
|
||||
self._multibyte_graph_cache_condition.notify_all()
|
||||
if stale is not None and not force_refresh:
|
||||
return stale
|
||||
raise
|
||||
|
||||
elapsed = time.monotonic() - started_at
|
||||
with self._multibyte_graph_cache_condition:
|
||||
self._multibyte_graph_cache_edges = computed
|
||||
self._multibyte_graph_cache_created_at = time.monotonic()
|
||||
self._multibyte_graph_cache_failure = None
|
||||
self._multibyte_graph_cache_failure_at = 0.0
|
||||
self._multibyte_graph_cache_computing = False
|
||||
self._multibyte_graph_cache_condition.notify_all()
|
||||
|
||||
self.logger.debug(
|
||||
"Computed %d multi-byte mesh edges in %.3fs",
|
||||
len(computed),
|
||||
elapsed,
|
||||
)
|
||||
return computed
|
||||
|
||||
def _compute_multibyte_evidence_edges(self) -> list[dict[str, Any]]:
|
||||
"""Derive mesh edges purely from multi-byte path evidence.
|
||||
|
||||
Splits each observed_paths row with bytes_per_hop >= 2 into consecutive
|
||||
@@ -2573,10 +2714,13 @@ class BotDataViewer:
|
||||
min_distance = request.args.get('min_distance', type=float)
|
||||
max_distance = request.args.get('max_distance', type=float)
|
||||
evidence = request.args.get('evidence', 'all')
|
||||
force_refresh = request.args.get('refresh') == '1'
|
||||
|
||||
if evidence == 'multibyte':
|
||||
edges, prefix_hex_chars = self._derive_multibyte_evidence_graph(
|
||||
days=days, min_observations=min_observations
|
||||
days=days,
|
||||
min_observations=min_observations,
|
||||
force_refresh=force_refresh,
|
||||
)
|
||||
return jsonify({
|
||||
'edges': edges,
|
||||
|
||||
@@ -648,8 +648,20 @@
|
||||
let isInitialMapLoad = true; // Track if this is the first time rendering the map with nodes
|
||||
let highlightedPath = null; // Currently highlighted path data
|
||||
let pathHighlightTimeout = null; // Debounce timer for path resolution
|
||||
let meshLoadInFlight = null;
|
||||
let pendingMeshLoad = null;
|
||||
let statsLoadInFlight = null;
|
||||
let meshLiveRefreshTimer = null;
|
||||
let meshLiveRefreshPending = false;
|
||||
let meshLiveRefreshNeedsNodes = false;
|
||||
let meshLiveRefreshRunning = false;
|
||||
let meshLiveRefreshRetryCount = 0;
|
||||
let meshLiveRefreshActiveReloadsNodes = false;
|
||||
|
||||
const PREFIX_HEX_CHARS = parseInt('{{ prefix_hex_chars|default(2) }}', 10);
|
||||
const MESH_LIVE_REFRESH_MS = 30000;
|
||||
const MESH_LIVE_REFRESH_RETRY_MS = 6000;
|
||||
const MESH_LIVE_REFRESH_MAX_RETRIES = 2;
|
||||
// Graph prefix length from edges API (2, 4, or 6 hex chars); used for node prefix display and node fetch.
|
||||
let graphPrefixHexChars = PREFIX_HEX_CHARS;
|
||||
// Format prefix for display: add byte count when multi-byte (2 or 3 bytes)
|
||||
@@ -1437,26 +1449,82 @@
|
||||
});
|
||||
});
|
||||
|
||||
// Load statistics
|
||||
async function loadStats() {
|
||||
function isValidMeshNode(node) {
|
||||
return Boolean(
|
||||
node
|
||||
&& typeof node === 'object'
|
||||
&& typeof node.public_key === 'string'
|
||||
&& node.public_key.length > 0
|
||||
&& typeof node.prefix === 'string'
|
||||
&& node.prefix.length > 0
|
||||
&& typeof node.name === 'string'
|
||||
&& Number.isFinite(node.latitude)
|
||||
&& Number.isFinite(node.longitude)
|
||||
);
|
||||
}
|
||||
|
||||
async function fetchMeshJson(url, requiredArrayKey = null) {
|
||||
const response = await fetch(url);
|
||||
let data;
|
||||
try {
|
||||
const response = await fetch('/api/mesh/stats');
|
||||
const stats = await response.json();
|
||||
|
||||
document.getElementById('stat-node-count').textContent = stats.node_count || 0;
|
||||
document.getElementById('stat-edge-count').textContent = stats.total_edges || 0;
|
||||
if (stats.multibyte_edges !== undefined) {
|
||||
document.getElementById('stat-edge-detail').textContent =
|
||||
`${stats.multibyte_edges} multi-byte confirmed`;
|
||||
}
|
||||
document.getElementById('stat-avg-obs').textContent = stats.avg_observations || 0;
|
||||
document.getElementById('stat-avg-dist').textContent = stats.avg_distance ? stats.avg_distance + ' km' : 'N/A';
|
||||
if (stats.bot_location) {
|
||||
botLocation = stats.bot_location; // used to frame the initial map view
|
||||
}
|
||||
data = await response.json();
|
||||
} catch (error) {
|
||||
console.error('Error loading stats:', error);
|
||||
throw new Error(`Invalid JSON response from ${url}`);
|
||||
}
|
||||
if (!response.ok || !data || typeof data !== 'object' || data.error) {
|
||||
const detail = data && data.error ? `: ${data.error}` : '';
|
||||
throw new Error(`Mesh request failed (${response.status})${detail}`);
|
||||
}
|
||||
if (requiredArrayKey && !Array.isArray(data[requiredArrayKey])) {
|
||||
throw new Error(
|
||||
`Mesh response from ${url} is missing ${requiredArrayKey}`
|
||||
);
|
||||
}
|
||||
if (
|
||||
requiredArrayKey === 'edges'
|
||||
&& !data.edges.every(edge =>
|
||||
edge
|
||||
&& typeof edge === 'object'
|
||||
&& typeof edge.from_prefix === 'string'
|
||||
&& typeof edge.to_prefix === 'string'
|
||||
)
|
||||
) {
|
||||
throw new Error(`Mesh response from ${url} contains invalid edges`);
|
||||
}
|
||||
if (
|
||||
requiredArrayKey === 'nodes'
|
||||
&& !data.nodes.every(isValidMeshNode)
|
||||
) {
|
||||
throw new Error(`Mesh response from ${url} contains invalid nodes`);
|
||||
}
|
||||
return data;
|
||||
}
|
||||
|
||||
// Load statistics
|
||||
function loadStats() {
|
||||
if (statsLoadInFlight) return statsLoadInFlight;
|
||||
statsLoadInFlight = (async () => {
|
||||
try {
|
||||
const stats = await fetchMeshJson('/api/mesh/stats');
|
||||
|
||||
document.getElementById('stat-node-count').textContent = stats.node_count || 0;
|
||||
document.getElementById('stat-edge-count').textContent = stats.total_edges || 0;
|
||||
if (stats.multibyte_edges !== undefined) {
|
||||
document.getElementById('stat-edge-detail').textContent =
|
||||
`${stats.multibyte_edges} multi-byte confirmed`;
|
||||
}
|
||||
document.getElementById('stat-avg-obs').textContent = stats.avg_observations || 0;
|
||||
document.getElementById('stat-avg-dist').textContent = stats.avg_distance ? stats.avg_distance + ' km' : 'N/A';
|
||||
if (stats.bot_location) {
|
||||
botLocation = stats.bot_location; // used to frame the initial map view
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Error loading stats:', error);
|
||||
}
|
||||
})().finally(() => {
|
||||
statsLoadInFlight = null;
|
||||
});
|
||||
return statsLoadInFlight;
|
||||
}
|
||||
|
||||
// --- Min Observations slider ---
|
||||
@@ -1504,13 +1572,64 @@
|
||||
saveFilters();
|
||||
}
|
||||
|
||||
// Load nodes and edges
|
||||
// Options may skip rendering or retain either already-loaded dataset.
|
||||
// Used by socket handlers so graph view is not re-rendered on live updates (avoids chaotic re-stabilization).
|
||||
async function loadData(options) {
|
||||
const skipRender = options && options.skipRender === true;
|
||||
const reloadEdges = !options || options.reloadEdges !== false;
|
||||
const reloadNodes = !options || options.reloadNodes !== false;
|
||||
function normalizeMeshLoadOptions(options) {
|
||||
return {
|
||||
skipRender: Boolean(options && options.skipRender === true),
|
||||
reloadEdges: !options || options.reloadEdges !== false,
|
||||
reloadNodes: !options || options.reloadNodes !== false,
|
||||
forceRefresh: Boolean(options && options.forceRefresh === true)
|
||||
};
|
||||
}
|
||||
|
||||
function mergeMeshLoadOptions(existing, incoming) {
|
||||
if (!existing) return incoming;
|
||||
return {
|
||||
// Render when any queued caller needs a render.
|
||||
skipRender: existing.skipRender && incoming.skipRender,
|
||||
reloadEdges: existing.reloadEdges || incoming.reloadEdges,
|
||||
reloadNodes: existing.reloadNodes || incoming.reloadNodes,
|
||||
forceRefresh: existing.forceRefresh || incoming.forceRefresh
|
||||
};
|
||||
}
|
||||
|
||||
// Load nodes and edges. Calls are serialized and requests that arrive while
|
||||
// one is active are merged into one trailing load.
|
||||
function loadData(options) {
|
||||
pendingMeshLoad = mergeMeshLoadOptions(
|
||||
pendingMeshLoad,
|
||||
normalizeMeshLoadOptions(options)
|
||||
);
|
||||
if (!meshLoadInFlight) {
|
||||
meshLoadInFlight = drainMeshLoads().finally(() => {
|
||||
meshLoadInFlight = null;
|
||||
});
|
||||
}
|
||||
return meshLoadInFlight;
|
||||
}
|
||||
|
||||
async function drainMeshLoads() {
|
||||
let lastError = null;
|
||||
while (pendingMeshLoad) {
|
||||
const options = pendingMeshLoad;
|
||||
pendingMeshLoad = null;
|
||||
try {
|
||||
await performMeshLoad(options);
|
||||
lastError = null;
|
||||
} catch (error) {
|
||||
// Continue with work queued while the failed request was active.
|
||||
lastError = error;
|
||||
}
|
||||
}
|
||||
if (lastError) {
|
||||
throw lastError;
|
||||
}
|
||||
}
|
||||
|
||||
async function performMeshLoad(options) {
|
||||
const skipRender = options.skipRender;
|
||||
const reloadEdges = options.reloadEdges;
|
||||
const reloadNodes = options.reloadNodes;
|
||||
const forceRefresh = options.forceRefresh;
|
||||
|
||||
try {
|
||||
const evidence = document.getElementById('filter-evidence').value;
|
||||
@@ -1519,6 +1638,7 @@
|
||||
const edgeParams = new URLSearchParams();
|
||||
if (evidence === 'multibyte') edgeParams.set('evidence', 'multibyte');
|
||||
if (edgeDays) edgeParams.set('days', edgeDays);
|
||||
if (forceRefresh) edgeParams.set('refresh', '1');
|
||||
const nodeParams = new URLSearchParams();
|
||||
if (nodeDays) nodeParams.set('days', nodeDays);
|
||||
const edgeQuery = edgeParams.toString();
|
||||
@@ -1533,43 +1653,60 @@
|
||||
// below once the edge prefix length is known, so the nodes request does
|
||||
// not have to wait for the edges response
|
||||
const [edgesData, nodesData] = await Promise.all([
|
||||
reloadEdges ? fetch(edgesUrl).then(r => r.json()) : Promise.resolve(null),
|
||||
reloadNodes ? fetch(nodesUrl).then(r => r.json()) : Promise.resolve(null)
|
||||
reloadEdges ? fetchMeshJson(edgesUrl, 'edges') : Promise.resolve(null),
|
||||
reloadNodes ? fetchMeshJson(nodesUrl, 'nodes') : Promise.resolve(null)
|
||||
]);
|
||||
if (edgesData) {
|
||||
allEdges = edgesData.edges || [];
|
||||
graphPrefixHexChars = (edgesData.prefix_hex_chars && [2, 4, 6].includes(edgesData.prefix_hex_chars))
|
||||
const nextEdges = edgesData ? edgesData.edges : allEdges;
|
||||
const nextPrefixHexChars = edgesData
|
||||
? ((edgesData.prefix_hex_chars && [2, 4, 6].includes(edgesData.prefix_hex_chars))
|
||||
? edgesData.prefix_hex_chars
|
||||
: PREFIX_HEX_CHARS;
|
||||
: PREFIX_HEX_CHARS)
|
||||
: graphPrefixHexChars;
|
||||
const sourceNodes = nodesData ? nodesData.nodes : allNodes;
|
||||
if (!nextEdges.every(edge =>
|
||||
edge
|
||||
&& typeof edge === 'object'
|
||||
&& typeof edge.from_prefix === 'string'
|
||||
&& typeof edge.to_prefix === 'string'
|
||||
)) {
|
||||
throw new Error(`Mesh response from ${edgesUrl} contains invalid edges`);
|
||||
}
|
||||
if (!sourceNodes.every(isValidMeshNode)) {
|
||||
throw new Error(`Mesh response from ${nodesUrl} contains invalid nodes`);
|
||||
}
|
||||
|
||||
if (nodesData) {
|
||||
allNodes = nodesData.nodes || [];
|
||||
}
|
||||
// Keep node.prefix in sync with edge prefixes (e.g. 2-byte graphs)
|
||||
allNodes.forEach(node => {
|
||||
// Prepare a complete candidate graph before replacing any current
|
||||
// state, so malformed responses cannot blank a working display.
|
||||
const nextNodes = sourceNodes.map(node => {
|
||||
const nextNode = { ...node };
|
||||
if (nextNode.public_key) {
|
||||
nextNode.prefix = nextNode.public_key
|
||||
.substring(0, nextPrefixHexChars)
|
||||
.toLowerCase();
|
||||
}
|
||||
return nextNode;
|
||||
});
|
||||
const nextNodeMap = {};
|
||||
const nextNodeMapByKey = {};
|
||||
nextNodes.forEach(node => {
|
||||
nextNodeMap[node.prefix] = node;
|
||||
if (node.public_key) {
|
||||
node.prefix = node.public_key.substring(0, graphPrefixHexChars).toLowerCase();
|
||||
nextNodeMapByKey[node.public_key] = node;
|
||||
}
|
||||
});
|
||||
|
||||
// Build node maps
|
||||
nodeMap = {};
|
||||
nodeMapByKey = {};
|
||||
allNodes.forEach(node => {
|
||||
nodeMap[node.prefix] = node; // Last node with this prefix wins (for backward compatibility)
|
||||
if (node.public_key) {
|
||||
nodeMapByKey[node.public_key] = node; // Map by public key for accurate lookups
|
||||
}
|
||||
});
|
||||
|
||||
// Build edge map
|
||||
edgeMap = {};
|
||||
allEdges.forEach(edge => {
|
||||
const nextEdgeMap = {};
|
||||
nextEdges.forEach(edge => {
|
||||
const key = `${edge.from_prefix}-${edge.to_prefix}`;
|
||||
edgeMap[key] = edge;
|
||||
nextEdgeMap[key] = edge;
|
||||
});
|
||||
|
||||
allEdges = nextEdges;
|
||||
allNodes = nextNodes;
|
||||
graphPrefixHexChars = nextPrefixHexChars;
|
||||
nodeMap = nextNodeMap;
|
||||
nodeMapByKey = nextNodeMapByKey;
|
||||
edgeMap = nextEdgeMap;
|
||||
|
||||
// Rescale the min-observations slider to the new edge distribution
|
||||
if (reloadEdges) {
|
||||
syncMinObsSlider();
|
||||
@@ -3164,9 +3301,111 @@
|
||||
return `${first8Bytes}...${last8Bytes}`;
|
||||
}
|
||||
|
||||
function refreshData() {
|
||||
loadStats();
|
||||
loadData();
|
||||
async function refreshData() {
|
||||
const absorbedPending = meshLiveRefreshPending
|
||||
|| meshLiveRefreshTimer !== null;
|
||||
const absorbedNeedsNodes = meshLiveRefreshNeedsNodes;
|
||||
if (meshLiveRefreshTimer) {
|
||||
clearTimeout(meshLiveRefreshTimer);
|
||||
meshLiveRefreshTimer = null;
|
||||
}
|
||||
meshLiveRefreshPending = false;
|
||||
meshLiveRefreshNeedsNodes = false;
|
||||
meshLiveRefreshRetryCount = 0;
|
||||
const activeScheduledRefresh = meshLiveRefreshRunning;
|
||||
const activeReloadsNodes = meshLiveRefreshActiveReloadsNodes;
|
||||
try {
|
||||
if (activeScheduledRefresh && meshLoadInFlight) {
|
||||
await Promise.all([loadStats(), meshLoadInFlight]);
|
||||
if (activeReloadsNodes) {
|
||||
applyFilters();
|
||||
} else {
|
||||
// The active forced edge refresh is sufficient; fetch only
|
||||
// the nodes it omitted and render from the combined state.
|
||||
await loadData({
|
||||
reloadEdges: false,
|
||||
reloadNodes: true
|
||||
});
|
||||
}
|
||||
} else {
|
||||
await Promise.all([
|
||||
loadStats(),
|
||||
loadData({ forceRefresh: true })
|
||||
]);
|
||||
}
|
||||
} catch (error) {
|
||||
// A manual refresh may have absorbed a pending live update. Restore
|
||||
// it and let the bounded scheduled-retry policy recover.
|
||||
scheduleMeshLiveRefresh(true, MESH_LIVE_REFRESH_RETRY_MS);
|
||||
throw error;
|
||||
}
|
||||
if (absorbedPending && activeScheduledRefresh) {
|
||||
// Events arriving during an active database snapshot may not be
|
||||
// represented by it, so retain that trailing refresh.
|
||||
scheduleMeshLiveRefresh(absorbedNeedsNodes);
|
||||
}
|
||||
}
|
||||
|
||||
function scheduleMeshLiveRefresh(
|
||||
reloadNodes,
|
||||
delayMs = MESH_LIVE_REFRESH_MS
|
||||
) {
|
||||
meshLiveRefreshPending = true;
|
||||
meshLiveRefreshNeedsNodes = meshLiveRefreshNeedsNodes || Boolean(reloadNodes);
|
||||
if (document.hidden || meshLiveRefreshTimer || meshLiveRefreshRunning) return;
|
||||
|
||||
meshLiveRefreshTimer = setTimeout(() => {
|
||||
meshLiveRefreshTimer = null;
|
||||
runScheduledMeshRefresh();
|
||||
}, delayMs);
|
||||
}
|
||||
|
||||
async function runScheduledMeshRefresh() {
|
||||
if (
|
||||
document.hidden
|
||||
|| meshLiveRefreshRunning
|
||||
|| !meshLiveRefreshPending
|
||||
) {
|
||||
return;
|
||||
}
|
||||
|
||||
const reloadNodes = meshLiveRefreshNeedsNodes;
|
||||
meshLiveRefreshPending = false;
|
||||
meshLiveRefreshNeedsNodes = false;
|
||||
meshLiveRefreshRunning = true;
|
||||
meshLiveRefreshActiveReloadsNodes = reloadNodes;
|
||||
let retryDelayMs = MESH_LIVE_REFRESH_MS;
|
||||
try {
|
||||
await Promise.all([
|
||||
loadStats(),
|
||||
loadData({
|
||||
reloadNodes,
|
||||
skipRender: true,
|
||||
forceRefresh: true
|
||||
})
|
||||
]);
|
||||
if (currentView === 'map') {
|
||||
applyFilters();
|
||||
}
|
||||
meshLiveRefreshRetryCount = 0;
|
||||
} catch (error) {
|
||||
console.error('Error refreshing live mesh data:', error);
|
||||
if (meshLiveRefreshRetryCount < MESH_LIVE_REFRESH_MAX_RETRIES) {
|
||||
meshLiveRefreshRetryCount++;
|
||||
meshLiveRefreshPending = true;
|
||||
meshLiveRefreshNeedsNodes =
|
||||
meshLiveRefreshNeedsNodes || reloadNodes;
|
||||
retryDelayMs = MESH_LIVE_REFRESH_RETRY_MS;
|
||||
} else {
|
||||
meshLiveRefreshRetryCount = 0;
|
||||
}
|
||||
} finally {
|
||||
meshLiveRefreshRunning = false;
|
||||
meshLiveRefreshActiveReloadsNodes = false;
|
||||
if (meshLiveRefreshPending) {
|
||||
scheduleMeshLiveRefresh(false, retryDelayMs);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function exportView() {
|
||||
@@ -3267,27 +3506,30 @@
|
||||
|
||||
// Socket.IO setup for real-time updates
|
||||
// Live updates are applied only when the map view is active. When the graph view is active,
|
||||
// we refresh data in the background but do not re-render the graph, to avoid the chaotic
|
||||
// re-stabilization layout (nodes clumping, edges tangling) that occurs when vis-network
|
||||
// runs physics again. User can refresh or switch to map and back to see updated graph.
|
||||
// events are coalesced into a bounded background refresh and do not re-render
|
||||
// the graph, avoiding both request storms and chaotic physics re-stabilization.
|
||||
document.addEventListener('visibilitychange', () => {
|
||||
if (!document.hidden && meshLiveRefreshPending) {
|
||||
if (meshLiveRefreshTimer) {
|
||||
clearTimeout(meshLiveRefreshTimer);
|
||||
meshLiveRefreshTimer = null;
|
||||
}
|
||||
runScheduledMeshRefresh();
|
||||
}
|
||||
});
|
||||
|
||||
function setupSocketIO() {
|
||||
const socket = window.connectionManager.socket;
|
||||
|
||||
socket.emit('subscribe_mesh');
|
||||
|
||||
function onMeshUpdate(data, label) {
|
||||
console.log(label, data);
|
||||
loadStats();
|
||||
loadData({ skipRender: true }).then(() => {
|
||||
if (currentView === 'map') {
|
||||
applyFilters();
|
||||
}
|
||||
});
|
||||
function onMeshUpdate(reloadNodes) {
|
||||
scheduleMeshLiveRefresh(reloadNodes);
|
||||
}
|
||||
|
||||
socket.on('mesh_edge_added', (data) => onMeshUpdate(data, 'New edge added:'));
|
||||
socket.on('mesh_edge_updated', (data) => onMeshUpdate(data, 'Edge updated:'));
|
||||
socket.on('mesh_node_added', (data) => onMeshUpdate(data, 'New node added:'));
|
||||
socket.on('mesh_edge_added', () => onMeshUpdate(false));
|
||||
socket.on('mesh_edge_updated', () => onMeshUpdate(false));
|
||||
socket.on('mesh_node_added', () => onMeshUpdate(true));
|
||||
}
|
||||
</script>
|
||||
|
||||
|
||||
@@ -3565,14 +3565,22 @@ class TestDbPathResolutionFromConfigDir:
|
||||
class TestWebViewerLoggingRespectsLogFile:
|
||||
"""Web viewer file logging follows [Logging] log_file like the main bot."""
|
||||
|
||||
def _write_config(self, config_dir: Path, log_file: str | None) -> str:
|
||||
def _write_config(
|
||||
self,
|
||||
config_dir: Path,
|
||||
log_file: str | None,
|
||||
log_level: str = "INFO",
|
||||
) -> str:
|
||||
cfg = configparser.ConfigParser()
|
||||
cfg["Connection"] = {"connection_type": "serial", "serial_port": "/dev/ttyUSB0"}
|
||||
cfg["Bot"] = {"bot_name": "TestBot", "db_path": "bot.db", "prefix_bytes": "1"}
|
||||
cfg["Channels"] = {"monitor_channels": "general"}
|
||||
cfg["Path_Command"] = {"max_hops": "5", "timeout": "30"}
|
||||
if log_file is not None:
|
||||
cfg["Logging"] = {"log_file": log_file}
|
||||
cfg["Logging"] = {
|
||||
"log_file": log_file,
|
||||
"log_level": log_level,
|
||||
}
|
||||
config_path = str(config_dir / "config.ini")
|
||||
with open(config_path, "w") as fh:
|
||||
cfg.write(fh)
|
||||
@@ -3614,6 +3622,60 @@ class TestWebViewerLoggingRespectsLogFile:
|
||||
assert not (config_dir / "logs").exists()
|
||||
assert not (tmp_path / "logs").exists()
|
||||
|
||||
def test_handlers_respect_configured_log_level(self, tmp_path: Path) -> None:
|
||||
import logging
|
||||
|
||||
config_dir = tmp_path / "cfg"
|
||||
log_dir = tmp_path / "varlog"
|
||||
config_dir.mkdir()
|
||||
log_dir.mkdir()
|
||||
config_path = self._write_config(
|
||||
config_dir,
|
||||
log_file=str(log_dir / "meshcore_bot.log"),
|
||||
log_level="WARNING",
|
||||
)
|
||||
|
||||
viewer = self._make_viewer(config_path)
|
||||
|
||||
assert viewer.logger.level == logging.WARNING
|
||||
assert viewer.logger.handlers
|
||||
assert all(
|
||||
handler.level == logging.WARNING
|
||||
for handler in viewer.logger.handlers
|
||||
)
|
||||
|
||||
def test_debug_file_logging_keeps_info_floor_for_journal(
|
||||
self, tmp_path: Path
|
||||
) -> None:
|
||||
import logging
|
||||
from logging.handlers import RotatingFileHandler
|
||||
|
||||
config_dir = tmp_path / "cfg"
|
||||
log_dir = tmp_path / "varlog"
|
||||
config_dir.mkdir()
|
||||
log_dir.mkdir()
|
||||
config_path = self._write_config(
|
||||
config_dir,
|
||||
log_file=str(log_dir / "meshcore_bot.log"),
|
||||
log_level="DEBUG",
|
||||
)
|
||||
|
||||
viewer = self._make_viewer(config_path)
|
||||
file_handlers = [
|
||||
handler
|
||||
for handler in viewer.logger.handlers
|
||||
if isinstance(handler, RotatingFileHandler)
|
||||
]
|
||||
console_handlers = [
|
||||
handler
|
||||
for handler in viewer.logger.handlers
|
||||
if not isinstance(handler, RotatingFileHandler)
|
||||
]
|
||||
|
||||
assert viewer.logger.level == logging.DEBUG
|
||||
assert [handler.level for handler in file_handlers] == [logging.DEBUG]
|
||||
assert [handler.level for handler in console_handlers] == [logging.INFO]
|
||||
|
||||
|
||||
class TestRadioDebugConfig:
|
||||
"""Tests for GET/POST /api/config/radio-debug endpoints."""
|
||||
|
||||
@@ -1,10 +1,14 @@
|
||||
"""Tests for modules.web_viewer.app — BotDataViewer Flask app."""
|
||||
|
||||
import json
|
||||
import shutil
|
||||
import sqlite3
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
from configparser import ConfigParser
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
@@ -625,6 +629,641 @@ class TestApiMeshEdgesEvidence:
|
||||
assert stats['multibyte_edges'] == 2
|
||||
|
||||
|
||||
class TestMultibyteMeshAggregateCache:
|
||||
def test_reuses_lifetime_aggregate_across_filtered_requests(
|
||||
self, viewer_with_db
|
||||
):
|
||||
_seed_observed_path(
|
||||
viewer_with_db.db_path,
|
||||
'aaaa11bbbb22',
|
||||
3,
|
||||
observation_count=5,
|
||||
)
|
||||
|
||||
with patch.object(
|
||||
viewer_with_db,
|
||||
'_compute_multibyte_evidence_edges',
|
||||
wraps=viewer_with_db._compute_multibyte_evidence_edges,
|
||||
) as compute:
|
||||
with viewer_with_db.app.test_client() as client:
|
||||
first = client.get('/api/mesh/edges?evidence=multibyte')
|
||||
second = client.get(
|
||||
'/api/mesh/edges?evidence=multibyte&min_observations=4&days=7'
|
||||
)
|
||||
|
||||
assert first.status_code == 200
|
||||
assert second.status_code == 200
|
||||
assert compute.call_count == 1
|
||||
|
||||
def test_recomputes_after_cache_window(self, viewer_with_db):
|
||||
with patch.object(
|
||||
viewer_with_db,
|
||||
'_compute_multibyte_evidence_edges',
|
||||
return_value=[],
|
||||
) as compute:
|
||||
viewer_with_db._aggregate_multibyte_evidence_edges()
|
||||
viewer_with_db._multibyte_graph_cache_created_at -= (
|
||||
viewer_with_db._mesh_graph_cache_seconds + 1
|
||||
)
|
||||
viewer_with_db._aggregate_multibyte_evidence_edges()
|
||||
|
||||
assert compute.call_count == 2
|
||||
|
||||
def test_forced_refresh_bypasses_warm_cache(self, viewer_with_db):
|
||||
_seed_observed_path(
|
||||
viewer_with_db.db_path,
|
||||
'aaaa11bbbb22',
|
||||
3,
|
||||
)
|
||||
with patch.object(
|
||||
viewer_with_db,
|
||||
'_compute_multibyte_evidence_edges',
|
||||
wraps=viewer_with_db._compute_multibyte_evidence_edges,
|
||||
) as compute:
|
||||
with viewer_with_db.app.test_client() as client:
|
||||
first = client.get('/api/mesh/edges?evidence=multibyte')
|
||||
_seed_observed_path(
|
||||
viewer_with_db.db_path,
|
||||
'cccc55dddd66',
|
||||
3,
|
||||
)
|
||||
cached = client.get('/api/mesh/edges?evidence=multibyte')
|
||||
refreshed = client.get(
|
||||
'/api/mesh/edges?evidence=multibyte&refresh=1'
|
||||
)
|
||||
|
||||
assert {
|
||||
edge['from_prefix'] for edge in first.get_json()['edges']
|
||||
} == {'aaaa11'}
|
||||
assert {
|
||||
edge['from_prefix'] for edge in cached.get_json()['edges']
|
||||
} == {'aaaa11'}
|
||||
assert {
|
||||
edge['from_prefix'] for edge in refreshed.get_json()['edges']
|
||||
} == {'aaaa11', 'cccc55'}
|
||||
assert compute.call_count == 2
|
||||
|
||||
def test_concurrent_cold_requests_share_one_computation(
|
||||
self, viewer_with_db
|
||||
):
|
||||
started = threading.Event()
|
||||
release = threading.Event()
|
||||
computed = [{'from_prefix': 'aaaa', 'to_prefix': 'bbbb'}]
|
||||
|
||||
def slow_compute():
|
||||
started.set()
|
||||
assert release.wait(timeout=2)
|
||||
return computed
|
||||
|
||||
results = []
|
||||
with patch.object(
|
||||
viewer_with_db,
|
||||
'_compute_multibyte_evidence_edges',
|
||||
side_effect=slow_compute,
|
||||
) as compute:
|
||||
first = threading.Thread(
|
||||
target=lambda: results.append(
|
||||
viewer_with_db._aggregate_multibyte_evidence_edges()
|
||||
)
|
||||
)
|
||||
second = threading.Thread(
|
||||
target=lambda: results.append(
|
||||
viewer_with_db._aggregate_multibyte_evidence_edges()
|
||||
)
|
||||
)
|
||||
first.start()
|
||||
assert started.wait(timeout=2)
|
||||
second.start()
|
||||
release.set()
|
||||
first.join(timeout=2)
|
||||
second.join(timeout=2)
|
||||
|
||||
assert not first.is_alive()
|
||||
assert not second.is_alive()
|
||||
assert compute.call_count == 1
|
||||
assert results == [computed, computed]
|
||||
|
||||
def test_concurrent_refresh_serves_stale_cache(self, viewer_with_db):
|
||||
stale = [{'from_prefix': 'old', 'to_prefix': 'edge'}]
|
||||
fresh = [{'from_prefix': 'new', 'to_prefix': 'edge'}]
|
||||
viewer_with_db._multibyte_graph_cache_edges = stale
|
||||
viewer_with_db._multibyte_graph_cache_created_at = 0
|
||||
started = threading.Event()
|
||||
release = threading.Event()
|
||||
|
||||
def slow_compute():
|
||||
started.set()
|
||||
assert release.wait(timeout=2)
|
||||
return fresh
|
||||
|
||||
refreshed = []
|
||||
with patch.object(
|
||||
viewer_with_db,
|
||||
'_compute_multibyte_evidence_edges',
|
||||
side_effect=slow_compute,
|
||||
) as compute:
|
||||
worker = threading.Thread(
|
||||
target=lambda: refreshed.append(
|
||||
viewer_with_db._aggregate_multibyte_evidence_edges()
|
||||
)
|
||||
)
|
||||
worker.start()
|
||||
assert started.wait(timeout=2)
|
||||
concurrent = viewer_with_db._aggregate_multibyte_evidence_edges()
|
||||
release.set()
|
||||
worker.join(timeout=2)
|
||||
|
||||
assert not worker.is_alive()
|
||||
assert concurrent is stale
|
||||
assert refreshed == [fresh]
|
||||
assert compute.call_count == 1
|
||||
|
||||
def test_concurrent_cold_failure_is_shared(self, viewer_with_db):
|
||||
started = threading.Event()
|
||||
release = threading.Event()
|
||||
errors = []
|
||||
|
||||
def failing_compute():
|
||||
started.set()
|
||||
assert release.wait(timeout=2)
|
||||
raise sqlite3.OperationalError('temporary failure')
|
||||
|
||||
def aggregate():
|
||||
try:
|
||||
viewer_with_db._aggregate_multibyte_evidence_edges()
|
||||
except Exception as exc:
|
||||
errors.append(exc)
|
||||
|
||||
with patch.object(
|
||||
viewer_with_db,
|
||||
'_compute_multibyte_evidence_edges',
|
||||
side_effect=failing_compute,
|
||||
) as compute:
|
||||
first = threading.Thread(target=aggregate)
|
||||
second = threading.Thread(target=aggregate)
|
||||
first.start()
|
||||
assert started.wait(timeout=2)
|
||||
second.start()
|
||||
release.set()
|
||||
first.join(timeout=2)
|
||||
second.join(timeout=2)
|
||||
|
||||
assert not first.is_alive()
|
||||
assert not second.is_alive()
|
||||
assert compute.call_count == 1
|
||||
assert len(errors) == 2
|
||||
assert any(isinstance(exc, sqlite3.OperationalError) for exc in errors)
|
||||
assert any(isinstance(exc, RuntimeError) for exc in errors)
|
||||
|
||||
def test_stale_failure_uses_retry_backoff(self, viewer_with_db):
|
||||
stale = [{'from_prefix': 'old', 'to_prefix': 'edge'}]
|
||||
viewer_with_db._multibyte_graph_cache_edges = stale
|
||||
viewer_with_db._multibyte_graph_cache_created_at = 0
|
||||
|
||||
with patch.object(
|
||||
viewer_with_db,
|
||||
'_compute_multibyte_evidence_edges',
|
||||
side_effect=sqlite3.OperationalError('temporary failure'),
|
||||
) as compute:
|
||||
first = viewer_with_db._aggregate_multibyte_evidence_edges()
|
||||
with pytest.raises(RuntimeError, match='retry suppressed'):
|
||||
viewer_with_db._aggregate_multibyte_evidence_edges(
|
||||
force_refresh=True
|
||||
)
|
||||
viewer_with_db._multibyte_graph_cache_failure_at -= (
|
||||
viewer_with_db._multibyte_graph_cache_retry_seconds + 1
|
||||
)
|
||||
third = viewer_with_db._aggregate_multibyte_evidence_edges()
|
||||
|
||||
assert first is stale
|
||||
assert third is stale
|
||||
assert compute.call_count == 2
|
||||
assert viewer_with_db._multibyte_graph_cache_failure == (
|
||||
'OperationalError',
|
||||
'temporary failure',
|
||||
)
|
||||
|
||||
def test_forced_failure_returns_500_while_normal_request_serves_stale(
|
||||
self, viewer_with_db
|
||||
):
|
||||
viewer_with_db._multibyte_graph_cache_edges = []
|
||||
viewer_with_db._multibyte_graph_cache_created_at = 0
|
||||
|
||||
with patch.object(
|
||||
viewer_with_db,
|
||||
'_compute_multibyte_evidence_edges',
|
||||
side_effect=sqlite3.OperationalError('temporary failure'),
|
||||
) as compute:
|
||||
with viewer_with_db.app.test_client() as client:
|
||||
forced = client.get(
|
||||
'/api/mesh/edges?evidence=multibyte&refresh=1'
|
||||
)
|
||||
normal = client.get('/api/mesh/edges?evidence=multibyte')
|
||||
|
||||
assert forced.status_code == 500
|
||||
assert forced.get_json()['error'] == 'An internal error occurred'
|
||||
assert normal.status_code == 200
|
||||
assert normal.get_json()['edges'] == []
|
||||
assert compute.call_count == 1
|
||||
|
||||
|
||||
def test_mesh_template_coalesces_live_refreshes():
|
||||
source = (
|
||||
Path(__file__).parents[1]
|
||||
/ 'modules'
|
||||
/ 'web_viewer'
|
||||
/ 'templates'
|
||||
/ 'mesh.html'
|
||||
).read_text(encoding='utf-8')
|
||||
|
||||
assert 'const MESH_LIVE_REFRESH_MS = 30000;' in source
|
||||
assert 'const MESH_LIVE_REFRESH_RETRY_MS = 6000;' in source
|
||||
assert "document.addEventListener('visibilitychange'" in source
|
||||
assert 'scheduleMeshLiveRefresh(reloadNodes);' in source
|
||||
assert "socket.on('mesh_edge_updated', () => onMeshUpdate(false));" in source
|
||||
assert 'while (pendingMeshLoad)' in source
|
||||
assert "fetchMeshJson(edgesUrl, 'edges')" in source
|
||||
assert "loadData({ skipRender: true }).then" not in source
|
||||
|
||||
|
||||
@pytest.mark.skipif(shutil.which('node') is None, reason='Node.js not installed')
|
||||
def test_mesh_refresh_coordinator_handles_bursts_visibility_and_failures():
|
||||
source = (
|
||||
Path(__file__).parents[1]
|
||||
/ 'modules'
|
||||
/ 'web_viewer'
|
||||
/ 'templates'
|
||||
/ 'mesh.html'
|
||||
).read_text(encoding='utf-8')
|
||||
|
||||
def extract(start, end):
|
||||
return source[source.index(start):source.index(end)]
|
||||
|
||||
coordinator = '\n'.join([
|
||||
extract(
|
||||
' function normalizeMeshLoadOptions',
|
||||
' function mergeMeshLoadOptions',
|
||||
),
|
||||
extract(
|
||||
' function mergeMeshLoadOptions',
|
||||
' // Load nodes and edges.',
|
||||
),
|
||||
extract(' function loadData', ' async function drainMeshLoads'),
|
||||
extract(
|
||||
' async function drainMeshLoads',
|
||||
' async function performMeshLoad',
|
||||
),
|
||||
extract(
|
||||
' async function refreshData',
|
||||
' function scheduleMeshLiveRefresh',
|
||||
),
|
||||
extract(
|
||||
' function scheduleMeshLiveRefresh',
|
||||
' async function runScheduledMeshRefresh',
|
||||
),
|
||||
extract(
|
||||
' async function runScheduledMeshRefresh',
|
||||
' function exportView',
|
||||
),
|
||||
])
|
||||
|
||||
script = f"""
|
||||
let meshLoadInFlight = null;
|
||||
let pendingMeshLoad = null;
|
||||
let meshLiveRefreshTimer = null;
|
||||
let meshLiveRefreshPending = false;
|
||||
let meshLiveRefreshNeedsNodes = false;
|
||||
let meshLiveRefreshRunning = false;
|
||||
let meshLiveRefreshRetryCount = 0;
|
||||
let meshLiveRefreshActiveReloadsNodes = false;
|
||||
const MESH_LIVE_REFRESH_MS = 10;
|
||||
const MESH_LIVE_REFRESH_RETRY_MS = 10;
|
||||
const MESH_LIVE_REFRESH_MAX_RETRIES = 2;
|
||||
let currentView = 'graph';
|
||||
let document = {{hidden: false}};
|
||||
let loadCount = 0;
|
||||
let statsCount = 0;
|
||||
let concurrent = 0;
|
||||
let maxConcurrent = 0;
|
||||
let forcedLoads = 0;
|
||||
let blockFirst = false;
|
||||
let failBlockedFirst = false;
|
||||
let firstStartedResolve;
|
||||
let firstReleaseResolve;
|
||||
let firstStarted = new Promise(resolve => firstStartedResolve = resolve);
|
||||
let firstRelease = new Promise(resolve => firstReleaseResolve = resolve);
|
||||
|
||||
async function performMeshLoad(options) {{
|
||||
loadCount++;
|
||||
if (options.forceRefresh) forcedLoads++;
|
||||
concurrent++;
|
||||
maxConcurrent = Math.max(maxConcurrent, concurrent);
|
||||
if (blockFirst && loadCount === 1) {{
|
||||
firstStartedResolve();
|
||||
await firstRelease;
|
||||
concurrent--;
|
||||
if (failBlockedFirst) {{
|
||||
throw new Error('expected first-load failure');
|
||||
}}
|
||||
return;
|
||||
}}
|
||||
await new Promise(resolve => setTimeout(resolve, 2));
|
||||
concurrent--;
|
||||
}}
|
||||
async function loadStats() {{ statsCount++; }}
|
||||
function applyFilters() {{}}
|
||||
|
||||
{coordinator}
|
||||
|
||||
(async () => {{
|
||||
for (let i = 0; i < 100; i++) scheduleMeshLiveRefresh(false);
|
||||
await new Promise(resolve => setTimeout(resolve, 40));
|
||||
const burst = {{loadCount, statsCount, maxConcurrent}};
|
||||
|
||||
loadCount = 0;
|
||||
statsCount = 0;
|
||||
maxConcurrent = 0;
|
||||
document.hidden = true;
|
||||
scheduleMeshLiveRefresh(false);
|
||||
await new Promise(resolve => setTimeout(resolve, 20));
|
||||
const hiddenLoads = loadCount;
|
||||
document.hidden = false;
|
||||
await runScheduledMeshRefresh();
|
||||
const visibleLoads = loadCount;
|
||||
|
||||
loadCount = 0;
|
||||
maxConcurrent = 0;
|
||||
blockFirst = true;
|
||||
failBlockedFirst = true;
|
||||
firstStarted = new Promise(resolve => firstStartedResolve = resolve);
|
||||
firstRelease = new Promise(resolve => firstReleaseResolve = resolve);
|
||||
const first = loadData({{reloadNodes: true}});
|
||||
await firstStarted;
|
||||
const trailing = loadData({{reloadNodes: false}});
|
||||
firstReleaseResolve();
|
||||
await Promise.all([first, trailing]);
|
||||
const failureQueue = {{
|
||||
loadCount,
|
||||
maxConcurrent,
|
||||
pending: pendingMeshLoad !== null
|
||||
}};
|
||||
|
||||
loadCount = 0;
|
||||
statsCount = 0;
|
||||
forcedLoads = 0;
|
||||
blockFirst = true;
|
||||
failBlockedFirst = true;
|
||||
firstStarted = new Promise(resolve => firstStartedResolve = resolve);
|
||||
firstRelease = new Promise(resolve => firstReleaseResolve = resolve);
|
||||
scheduleMeshLiveRefresh(true, 10);
|
||||
await firstStarted;
|
||||
firstReleaseResolve();
|
||||
await new Promise(resolve => setTimeout(resolve, 35));
|
||||
const scheduledRetry = {{
|
||||
loadCount,
|
||||
statsCount,
|
||||
forcedLoads,
|
||||
pending: meshLiveRefreshPending,
|
||||
retries: meshLiveRefreshRetryCount
|
||||
}};
|
||||
|
||||
loadCount = 0;
|
||||
statsCount = 0;
|
||||
forcedLoads = 0;
|
||||
blockFirst = false;
|
||||
failBlockedFirst = false;
|
||||
scheduleMeshLiveRefresh(false, 20);
|
||||
await refreshData();
|
||||
await new Promise(resolve => setTimeout(resolve, 30));
|
||||
const manualAbsorbsTimer = {{
|
||||
loadCount,
|
||||
statsCount,
|
||||
forcedLoads,
|
||||
pending: meshLiveRefreshPending
|
||||
}};
|
||||
|
||||
loadCount = 0;
|
||||
statsCount = 0;
|
||||
forcedLoads = 0;
|
||||
blockFirst = true;
|
||||
failBlockedFirst = true;
|
||||
firstStarted = new Promise(resolve => firstStartedResolve = resolve);
|
||||
firstRelease = new Promise(resolve => firstReleaseResolve = resolve);
|
||||
scheduleMeshLiveRefresh(false, 20);
|
||||
const failedManualPromise = refreshData().catch(() => {{}});
|
||||
await firstStarted;
|
||||
firstReleaseResolve();
|
||||
await failedManualPromise;
|
||||
await new Promise(resolve => setTimeout(resolve, 35));
|
||||
const failedManualRetries = {{
|
||||
loadCount,
|
||||
forcedLoads,
|
||||
pending: meshLiveRefreshPending,
|
||||
retries: meshLiveRefreshRetryCount
|
||||
}};
|
||||
|
||||
loadCount = 0;
|
||||
statsCount = 0;
|
||||
forcedLoads = 0;
|
||||
blockFirst = true;
|
||||
failBlockedFirst = false;
|
||||
firstStarted = new Promise(resolve => firstStartedResolve = resolve);
|
||||
firstRelease = new Promise(resolve => firstReleaseResolve = resolve);
|
||||
scheduleMeshLiveRefresh(true, 0);
|
||||
await firstStarted;
|
||||
const activeManualPromise = refreshData();
|
||||
firstReleaseResolve();
|
||||
await activeManualPromise;
|
||||
await new Promise(resolve => setTimeout(resolve, 10));
|
||||
const manualReusesActiveRefresh = {{
|
||||
loadCount,
|
||||
forcedLoads,
|
||||
pending: meshLiveRefreshPending
|
||||
}};
|
||||
|
||||
console.log(JSON.stringify({{
|
||||
burst,
|
||||
hiddenLoads,
|
||||
visibleLoads,
|
||||
failureQueue,
|
||||
scheduledRetry,
|
||||
manualAbsorbsTimer,
|
||||
failedManualRetries,
|
||||
manualReusesActiveRefresh
|
||||
}}));
|
||||
}})().catch(error => {{
|
||||
console.error(error);
|
||||
process.exitCode = 1;
|
||||
}});
|
||||
"""
|
||||
|
||||
completed = subprocess.run(
|
||||
[shutil.which('node'), '-'],
|
||||
input=script,
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=True,
|
||||
timeout=5,
|
||||
)
|
||||
result = json.loads(completed.stdout.strip())
|
||||
|
||||
assert result['burst'] == {
|
||||
'loadCount': 1,
|
||||
'statsCount': 1,
|
||||
'maxConcurrent': 1,
|
||||
}
|
||||
assert result['hiddenLoads'] == 0
|
||||
assert result['visibleLoads'] == 1
|
||||
assert result['failureQueue'] == {
|
||||
'loadCount': 2,
|
||||
'maxConcurrent': 1,
|
||||
'pending': False,
|
||||
}
|
||||
assert result['scheduledRetry'] == {
|
||||
'loadCount': 2,
|
||||
'statsCount': 2,
|
||||
'forcedLoads': 2,
|
||||
'pending': False,
|
||||
'retries': 0,
|
||||
}
|
||||
assert result['manualAbsorbsTimer'] == {
|
||||
'loadCount': 1,
|
||||
'statsCount': 1,
|
||||
'forcedLoads': 1,
|
||||
'pending': False,
|
||||
}
|
||||
assert result['failedManualRetries'] == {
|
||||
'loadCount': 2,
|
||||
'forcedLoads': 2,
|
||||
'pending': False,
|
||||
'retries': 0,
|
||||
}
|
||||
assert result['manualReusesActiveRefresh'] == {
|
||||
'loadCount': 1,
|
||||
'forcedLoads': 1,
|
||||
'pending': False,
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.skipif(shutil.which('node') is None, reason='Node.js not installed')
|
||||
def test_mesh_fetch_rejects_http_errors_and_invalid_payloads():
|
||||
source = (
|
||||
Path(__file__).parents[1]
|
||||
/ 'modules'
|
||||
/ 'web_viewer'
|
||||
/ 'templates'
|
||||
/ 'mesh.html'
|
||||
).read_text(encoding='utf-8')
|
||||
helper = source[
|
||||
source.index(' function isValidMeshNode'):
|
||||
source.index(' // Load statistics')
|
||||
]
|
||||
script = f"""
|
||||
{helper}
|
||||
(async () => {{
|
||||
global.fetch = async () => ({{
|
||||
ok: false,
|
||||
status: 500,
|
||||
json: async () => ({{error: 'temporary failure'}})
|
||||
}});
|
||||
let httpError = '';
|
||||
try {{
|
||||
await fetchMeshJson('/api/mesh/edges', 'edges');
|
||||
}} catch (error) {{
|
||||
httpError = error.message;
|
||||
}}
|
||||
|
||||
global.fetch = async () => ({{
|
||||
ok: true,
|
||||
status: 200,
|
||||
json: async () => ({{edges: null}})
|
||||
}});
|
||||
let shapeError = '';
|
||||
try {{
|
||||
await fetchMeshJson('/api/mesh/edges', 'edges');
|
||||
}} catch (error) {{
|
||||
shapeError = error.message;
|
||||
}}
|
||||
|
||||
global.fetch = async () => ({{
|
||||
ok: true,
|
||||
status: 200,
|
||||
json: async () => ({{edges: [null]}})
|
||||
}});
|
||||
let nestedShapeError = '';
|
||||
try {{
|
||||
await fetchMeshJson('/api/mesh/edges', 'edges');
|
||||
}} catch (error) {{
|
||||
nestedShapeError = error.message;
|
||||
}}
|
||||
|
||||
global.fetch = async () => ({{
|
||||
ok: true,
|
||||
status: 200,
|
||||
json: async () => ({{nodes: [{{}}]}})
|
||||
}});
|
||||
let emptyNodeError = '';
|
||||
try {{
|
||||
await fetchMeshJson('/api/mesh/nodes', 'nodes');
|
||||
}} catch (error) {{
|
||||
emptyNodeError = error.message;
|
||||
}}
|
||||
|
||||
global.fetch = async () => ({{
|
||||
ok: true,
|
||||
status: 200,
|
||||
json: async () => ({{nodes: [{{
|
||||
public_key: 'aabb',
|
||||
prefix: 'aa',
|
||||
name: 'Node',
|
||||
latitude: '1.0',
|
||||
longitude: 2.0
|
||||
}}]}})
|
||||
}});
|
||||
let typedNodeError = '';
|
||||
try {{
|
||||
await fetchMeshJson('/api/mesh/nodes', 'nodes');
|
||||
}} catch (error) {{
|
||||
typedNodeError = error.message;
|
||||
}}
|
||||
console.log(JSON.stringify({{
|
||||
httpError,
|
||||
shapeError,
|
||||
nestedShapeError,
|
||||
emptyNodeError,
|
||||
typedNodeError
|
||||
}}));
|
||||
}})().catch(error => {{
|
||||
console.error(error);
|
||||
process.exitCode = 1;
|
||||
}});
|
||||
"""
|
||||
completed = subprocess.run(
|
||||
[shutil.which('node'), '-'],
|
||||
input=script,
|
||||
text=True,
|
||||
capture_output=True,
|
||||
check=True,
|
||||
timeout=5,
|
||||
)
|
||||
result = json.loads(completed.stdout.strip())
|
||||
assert result['httpError'] == (
|
||||
'Mesh request failed (500): temporary failure'
|
||||
)
|
||||
assert result['shapeError'] == (
|
||||
'Mesh response from /api/mesh/edges is missing edges'
|
||||
)
|
||||
assert result['nestedShapeError'] == (
|
||||
'Mesh response from /api/mesh/edges contains invalid edges'
|
||||
)
|
||||
assert result['emptyNodeError'] == (
|
||||
'Mesh response from /api/mesh/nodes contains invalid nodes'
|
||||
)
|
||||
assert result['typedNodeError'] == (
|
||||
'Mesh response from /api/mesh/nodes contains invalid nodes'
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# api_geocode_contact
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user