mirror of
https://github.com/Quad4-Software/MeshChatX.git
synced 2026-08-18 09:49:09 -04:00
feat(memory-pressure): implement memory management features including periodic cleanup and SQLite pragmas for low memory conditions
This commit is contained in:
Ivan
parent
08f98b9877
commit
8b30976980
19 changed files with 1050 additions and 17 deletions
|
|
@ -9,7 +9,7 @@ const {
|
|||
loadCrashReport,
|
||||
persistCrashReport,
|
||||
} = require("./backendCrashReport");
|
||||
const { killOrphanBackendProcesses } = require("./backendProcessWin");
|
||||
const { killOrphanBackendProcesses } = require("./backendProcessOrphans");
|
||||
|
||||
const LOG_LINE_CAP = 200;
|
||||
|
||||
|
|
|
|||
128
electron/backendProcessOrphans.js
Normal file
128
electron/backendProcessOrphans.js
Normal file
|
|
@ -0,0 +1,128 @@
|
|||
"use strict";
|
||||
|
||||
const { execFileSync } = require("node:child_process");
|
||||
|
||||
const { killOrphanBackendProcesses: killOrphanBackendProcessesWin } = require("./backendProcessWin");
|
||||
|
||||
const BACKEND_IMAGE_UNIX = "ReticulumMeshChatX";
|
||||
|
||||
/**
|
||||
* @param {string} args
|
||||
* @returns {boolean}
|
||||
*/
|
||||
function isHeadlessBackendArgs(args) {
|
||||
if (typeof args !== "string" || !args) {
|
||||
return false;
|
||||
}
|
||||
if (!args.includes(BACKEND_IMAGE_UNIX)) {
|
||||
return false;
|
||||
}
|
||||
return /\s--headless(?:\s|$)/.test(args) || args.endsWith(" --headless");
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {number|null|undefined} ownPid
|
||||
* @returns {number[]}
|
||||
*/
|
||||
function listUnixBackendPids(ownPid = null) {
|
||||
if (process.platform === "win32") {
|
||||
return [];
|
||||
}
|
||||
try {
|
||||
const out = execFileSync("ps", ["-eo", "pid=,args="], {
|
||||
encoding: "utf8",
|
||||
maxBuffer: 8 * 1024 * 1024,
|
||||
});
|
||||
const pids = [];
|
||||
for (const line of out.split(/\n/)) {
|
||||
const trimmed = line.trim();
|
||||
if (!trimmed) {
|
||||
continue;
|
||||
}
|
||||
const match = trimmed.match(/^(\d+)\s+(.*)$/);
|
||||
if (!match) {
|
||||
continue;
|
||||
}
|
||||
const pid = Number(match[1]);
|
||||
if (!Number.isFinite(pid) || pid <= 0) {
|
||||
continue;
|
||||
}
|
||||
if (ownPid != null && pid === ownPid) {
|
||||
continue;
|
||||
}
|
||||
if (pid === process.pid) {
|
||||
continue;
|
||||
}
|
||||
if (!isHeadlessBackendArgs(match[2])) {
|
||||
continue;
|
||||
}
|
||||
pids.push(pid);
|
||||
}
|
||||
return pids;
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {number[]} pids
|
||||
*/
|
||||
function killUnixPids(pids) {
|
||||
if (!Array.isArray(pids)) {
|
||||
return;
|
||||
}
|
||||
for (const pid of pids) {
|
||||
try {
|
||||
process.kill(pid, "SIGTERM");
|
||||
} catch {
|
||||
/* process may already be gone */
|
||||
}
|
||||
}
|
||||
const deadline = Date.now() + 1500;
|
||||
while (Date.now() < deadline) {
|
||||
let remaining = 0;
|
||||
for (const pid of pids) {
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
remaining += 1;
|
||||
} catch {
|
||||
/* gone */
|
||||
}
|
||||
}
|
||||
if (remaining === 0) {
|
||||
return;
|
||||
}
|
||||
const spinUntil = Date.now() + 50;
|
||||
while (Date.now() < spinUntil) {
|
||||
/* brief wait for SIGTERM to take effect */
|
||||
}
|
||||
}
|
||||
for (const pid of pids) {
|
||||
try {
|
||||
process.kill(pid, "SIGKILL");
|
||||
} catch {
|
||||
/* process may already be gone */
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @param {number|null|undefined} ownPid
|
||||
* @returns {number}
|
||||
*/
|
||||
function killOrphanBackendProcesses(ownPid = null) {
|
||||
if (process.platform === "win32") {
|
||||
return killOrphanBackendProcessesWin(ownPid);
|
||||
}
|
||||
const pids = listUnixBackendPids(ownPid);
|
||||
killUnixPids(pids);
|
||||
return pids.length;
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
BACKEND_IMAGE_UNIX,
|
||||
isHeadlessBackendArgs,
|
||||
killOrphanBackendProcesses,
|
||||
killUnixPids,
|
||||
listUnixBackendPids,
|
||||
};
|
||||
|
|
@ -139,11 +139,11 @@ from meshchatx.src.backend.meshchat_utils import (
|
|||
parse_lxmf_stamp_cost,
|
||||
parse_nomadnetwork_node_display_name,
|
||||
)
|
||||
from meshchatx.src.backend.memory_pressure import MemoryPressureManager, cache_stats
|
||||
from meshchatx.src.backend.nomadnet_downloader import (
|
||||
NomadnetFileDownloader,
|
||||
NomadnetPageDownloader,
|
||||
get_cached_active_link,
|
||||
sweep_stale_links,
|
||||
)
|
||||
from meshchatx.src.backend.nomadnet_utils import (
|
||||
convert_nomadnet_field_data_to_map,
|
||||
|
|
@ -514,6 +514,7 @@ class ReticulumMeshChat:
|
|||
reticulum_getter=lambda: getattr(self, "reticulum", None),
|
||||
broadcast_event=self._on_rns_link_broadcast,
|
||||
)
|
||||
self.memory_pressure = MemoryPressureManager(app=self)
|
||||
# Track long-running rns.link.* handler tasks per WS client so they can
|
||||
# be cancelled when the client disconnects.
|
||||
self._rns_link_tasks: dict[web.WebSocketResponse, set[asyncio.Task]] = {}
|
||||
|
|
@ -3048,7 +3049,10 @@ class ReticulumMeshChat:
|
|||
gc_counter += 1
|
||||
if gc_counter >= 300:
|
||||
gc_counter = 0
|
||||
sweep_stale_links()
|
||||
try:
|
||||
await asyncio.to_thread(self.memory_pressure.run_periodic_cleanup)
|
||||
except Exception as exc:
|
||||
print(f"[memory_pressure] periodic cleanup error: {exc}")
|
||||
# Python 3.14+ incremental GC: with threshold[2]==0 full gen2
|
||||
# collections are never scheduled automatically, so force one.
|
||||
if sys.version_info >= (3, 14) and gc.get_threshold()[2] == 0:
|
||||
|
|
@ -7217,6 +7221,18 @@ class ReticulumMeshChat:
|
|||
None,
|
||||
),
|
||||
"busy_timeout": _safe_sqlite_pragma("busy_timeout", None),
|
||||
"temp_store": _safe_sqlite_pragma("temp_store", None),
|
||||
"cache_size": _safe_sqlite_pragma("cache_size", None),
|
||||
"mmap_size": _safe_sqlite_pragma("mmap_size", None),
|
||||
"memory_relaxed": bool(
|
||||
getattr(
|
||||
self.database,
|
||||
"_sqlite_memory_relaxed",
|
||||
False,
|
||||
)
|
||||
if self.database is not None
|
||||
else False
|
||||
),
|
||||
},
|
||||
"reticulum_config_path": self._api_reticulum_config_path(),
|
||||
"host_platform": sys.platform,
|
||||
|
|
@ -7243,6 +7259,12 @@ class ReticulumMeshChat:
|
|||
"announces_per_second": announces_per_second,
|
||||
"announces_per_minute": announces_per_minute,
|
||||
"announces_per_hour": announces_per_hour,
|
||||
**cache_stats(),
|
||||
"memory_cleanup": getattr(
|
||||
getattr(self, "memory_pressure", None),
|
||||
"last_stats",
|
||||
{},
|
||||
),
|
||||
},
|
||||
"is_reticulum_running": hasattr(self, "reticulum")
|
||||
and self.reticulum is not None,
|
||||
|
|
|
|||
|
|
@ -87,6 +87,7 @@ class Database:
|
|||
self.debug_logs = DebugLogsDAO(self.provider)
|
||||
self.access_attempts = AccessAttemptsDAO(self.provider)
|
||||
self.crash_history = CrashHistoryDAO(self.provider)
|
||||
self._sqlite_memory_relaxed = False
|
||||
|
||||
def initialize(self):
|
||||
self._tune_sqlite_pragmas()
|
||||
|
|
@ -104,9 +105,28 @@ class Database:
|
|||
self.execute_sql("PRAGMA cache_size=-8000") # 8 MB
|
||||
self.execute_sql("PRAGMA mmap_size=67108864") # 64 MB
|
||||
self.execute_sql("PRAGMA busy_timeout=5000") # 5 s wait on lock contention
|
||||
self._sqlite_memory_relaxed = False
|
||||
except Exception as exc:
|
||||
print(f"SQLite pragma setup failed: {exc}")
|
||||
|
||||
def apply_memory_pressure_pragmas(self, relax: bool) -> bool:
|
||||
"""Move SQLite temp/cache work toward disk when host RAM is low."""
|
||||
try:
|
||||
if relax:
|
||||
self.execute_sql("PRAGMA temp_store=FILE")
|
||||
self.execute_sql("PRAGMA cache_size=-2000") # 2 MB
|
||||
self.execute_sql("PRAGMA mmap_size=0")
|
||||
self._sqlite_memory_relaxed = True
|
||||
else:
|
||||
self.execute_sql("PRAGMA temp_store=MEMORY")
|
||||
self.execute_sql("PRAGMA cache_size=-8000")
|
||||
self.execute_sql("PRAGMA mmap_size=67108864")
|
||||
self._sqlite_memory_relaxed = False
|
||||
return True
|
||||
except Exception as exc:
|
||||
print(f"SQLite memory-pressure pragma update failed: {exc}")
|
||||
return False
|
||||
|
||||
def _get_pragma_value(self, pragma: str, default=None):
|
||||
safe = _sanitize_pragma_read_name(pragma)
|
||||
if safe is None:
|
||||
|
|
|
|||
164
meshchatx/src/backend/memory_pressure.py
Normal file
164
meshchatx/src/backend/memory_pressure.py
Normal file
|
|
@ -0,0 +1,164 @@
|
|||
# SPDX-License-Identifier: 0BSD
|
||||
|
||||
"""Periodic and pressure-triggered memory cleanup for MeshChatX."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import time
|
||||
from typing import Any
|
||||
|
||||
from meshchatx.src.backend import nomadnet_downloader, reticulum_pathfinding, rns_link_manager
|
||||
|
||||
_log = logging.getLogger("meshchatx.memory_pressure")
|
||||
|
||||
HELD_ANNOUNCES_DROP_THRESHOLD = 512
|
||||
ANNOUNCE_CACHE_CLEAN_INTERVAL_S = 30 * 60
|
||||
|
||||
|
||||
class MemoryPressureManager:
|
||||
"""Coordinates link sweeps, path pruning, and SQLite disk offload."""
|
||||
|
||||
def __init__(self, app: Any = None):
|
||||
self.app = app
|
||||
self._last_announce_cache_clean = 0.0
|
||||
self._sqlite_relaxed = False
|
||||
self.last_stats: dict[str, Any] = {
|
||||
"nomad_links_swept": 0,
|
||||
"rns_links_swept": 0,
|
||||
"paths_pruned_expired": 0,
|
||||
"paths_pruned_cap": 0,
|
||||
"announce_cache_cleaned": False,
|
||||
"held_queues_dropped": False,
|
||||
"sqlite_relaxed": False,
|
||||
}
|
||||
|
||||
def run_periodic_cleanup(self) -> dict[str, Any]:
|
||||
"""Called from announce_loop about every 5 minutes."""
|
||||
reticulum = getattr(self.app, "reticulum", None) if self.app else None
|
||||
|
||||
before_nomad = nomadnet_downloader.cached_link_count()
|
||||
nomadnet_downloader.sweep_stale_links()
|
||||
after_nomad = nomadnet_downloader.cached_link_count()
|
||||
|
||||
before_rns = rns_link_manager.cached_link_count()
|
||||
rns_link_manager.sweep_stale_links()
|
||||
after_rns = rns_link_manager.cached_link_count()
|
||||
|
||||
expired_dropped = reticulum_pathfinding.prune_expired_path_table_entries(
|
||||
reticulum,
|
||||
)
|
||||
cap_dropped = reticulum_pathfinding.prune_path_table_to_soft_cap(reticulum)
|
||||
|
||||
announce_cleaned = False
|
||||
now = time.time()
|
||||
if now - self._last_announce_cache_clean >= ANNOUNCE_CACHE_CLEAN_INTERVAL_S:
|
||||
announce_cleaned = reticulum_pathfinding.clean_rns_announce_cache()
|
||||
if announce_cleaned:
|
||||
self._last_announce_cache_clean = now
|
||||
|
||||
held_dropped = self._maybe_drop_held_announce_queues()
|
||||
|
||||
self.last_stats = {
|
||||
"nomad_links_swept": max(0, before_nomad - after_nomad),
|
||||
"rns_links_swept": max(0, before_rns - after_rns),
|
||||
"nomad_cached_links": after_nomad,
|
||||
"rns_cached_links": after_rns,
|
||||
"paths_pruned_expired": expired_dropped,
|
||||
"paths_pruned_cap": cap_dropped,
|
||||
"path_table_size": reticulum_pathfinding.path_table_size(),
|
||||
"announce_cache_cleaned": announce_cleaned,
|
||||
"held_queues_dropped": held_dropped,
|
||||
"sqlite_relaxed": self._sqlite_relaxed,
|
||||
}
|
||||
if (
|
||||
expired_dropped
|
||||
or cap_dropped
|
||||
or announce_cleaned
|
||||
or held_dropped
|
||||
or before_nomad != after_nomad
|
||||
or before_rns != after_rns
|
||||
):
|
||||
_log.info("Memory cleanup: %s", self.last_stats)
|
||||
return self.last_stats
|
||||
|
||||
def on_memory_low(self, available_mb: float) -> dict[str, Any]:
|
||||
"""Reactive cleanup when HealthMonitor reports low available RAM."""
|
||||
stats = self.run_periodic_cleanup()
|
||||
db = getattr(self.app, "database", None) if self.app else None
|
||||
if db is not None and hasattr(db, "apply_memory_pressure_pragmas"):
|
||||
try:
|
||||
db.apply_memory_pressure_pragmas(True)
|
||||
self._sqlite_relaxed = True
|
||||
stats["sqlite_relaxed"] = True
|
||||
except Exception as exc:
|
||||
_log.debug("SQLite pressure pragmas failed: %s", exc)
|
||||
_log.warning(
|
||||
"Memory pressure cleanup (available=%.0f MB): %s",
|
||||
available_mb,
|
||||
stats,
|
||||
)
|
||||
return stats
|
||||
|
||||
def on_memory_recovered(self) -> None:
|
||||
if not self._sqlite_relaxed:
|
||||
return
|
||||
db = getattr(self.app, "database", None) if self.app else None
|
||||
if db is not None and hasattr(db, "apply_memory_pressure_pragmas"):
|
||||
try:
|
||||
db.apply_memory_pressure_pragmas(False)
|
||||
self._sqlite_relaxed = False
|
||||
self.last_stats["sqlite_relaxed"] = False
|
||||
except Exception as exc:
|
||||
_log.debug("SQLite restore pragmas failed: %s", exc)
|
||||
|
||||
def _maybe_drop_held_announce_queues(self) -> bool:
|
||||
if self.app is None:
|
||||
return False
|
||||
handler = getattr(self.app, "rnpath_handler", None)
|
||||
if handler is None or not hasattr(handler, "drop_announce_queues"):
|
||||
return False
|
||||
held = self._count_held_announces()
|
||||
if held < HELD_ANNOUNCES_DROP_THRESHOLD:
|
||||
return False
|
||||
try:
|
||||
handler.drop_announce_queues()
|
||||
_log.info("Dropped announce queues (held_announces=%s)", held)
|
||||
return True
|
||||
except Exception as exc:
|
||||
_log.debug("drop_announce_queues failed: %s", exc)
|
||||
return False
|
||||
|
||||
def _count_held_announces(self) -> int:
|
||||
total = 0
|
||||
try:
|
||||
held = getattr(
|
||||
__import__("RNS").Transport,
|
||||
"held_announces",
|
||||
None,
|
||||
)
|
||||
if isinstance(held, dict):
|
||||
total += len(held)
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
import RNS
|
||||
|
||||
interfaces = getattr(RNS.Transport, "interfaces", None) or []
|
||||
for iface in interfaces:
|
||||
iface_held = getattr(iface, "held_announces", None)
|
||||
if isinstance(iface_held, dict):
|
||||
total += len(iface_held)
|
||||
elif isinstance(iface_held, (list, set, tuple)):
|
||||
total += len(iface_held)
|
||||
except Exception:
|
||||
pass
|
||||
return total
|
||||
|
||||
|
||||
def cache_stats() -> dict[str, int]:
|
||||
return {
|
||||
"nomad_cached_links": nomadnet_downloader.cached_link_count(),
|
||||
"rns_cached_links": rns_link_manager.cached_link_count(),
|
||||
"path_table_size": reticulum_pathfinding.path_table_size(),
|
||||
}
|
||||
|
|
@ -15,12 +15,22 @@ from meshchatx.src.backend.reticulum_pathfinding import ReticulumLike
|
|||
# Global cache for Nomad Network links (reuse instead of reconnecting per request).
|
||||
# Protected by _nomadnet_links_lock for callers that may touch Reticulum from multiple threads.
|
||||
nomadnet_cached_links: dict[bytes, object] = {}
|
||||
_nomadnet_link_last_used: dict[bytes, float] = {}
|
||||
_nomadnet_links_lock = threading.Lock()
|
||||
|
||||
# Cap active Nomad links retained in process memory.
|
||||
MAX_CACHED_LINKS = 32
|
||||
LINK_IDLE_TTL_S = 30 * 60
|
||||
|
||||
# Wait granularity while polling for path / link (seconds). Smaller = faster reaction, slightly more wakeups.
|
||||
_POLL_INTERVAL_S = 0.02
|
||||
|
||||
|
||||
def cached_link_count() -> int:
|
||||
with _nomadnet_links_lock:
|
||||
return len(nomadnet_cached_links)
|
||||
|
||||
|
||||
def get_cached_active_link(destination_hash: bytes):
|
||||
"""Return a cached link if present and ACTIVE; drop stale entries."""
|
||||
with _nomadnet_links_lock:
|
||||
|
|
@ -28,16 +38,30 @@ def get_cached_active_link(destination_hash: bytes):
|
|||
if link is None:
|
||||
return None
|
||||
if link.status is RNS.Link.ACTIVE:
|
||||
_nomadnet_link_last_used[destination_hash] = time.time()
|
||||
return link
|
||||
try:
|
||||
del nomadnet_cached_links[destination_hash]
|
||||
except KeyError:
|
||||
pass
|
||||
_nomadnet_link_last_used.pop(destination_hash, None)
|
||||
return None
|
||||
|
||||
|
||||
def _teardown_links(links) -> None:
|
||||
for link in links:
|
||||
if link is None:
|
||||
continue
|
||||
try:
|
||||
link.teardown()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def sweep_stale_links():
|
||||
"""Evict all non-ACTIVE links from the global cache."""
|
||||
"""Evict non-ACTIVE, idle, and over-cap links from the global cache."""
|
||||
now = time.time()
|
||||
to_teardown = []
|
||||
with _nomadnet_links_lock:
|
||||
stale = [
|
||||
k
|
||||
|
|
@ -45,14 +69,52 @@ def sweep_stale_links():
|
|||
if v.status is not RNS.Link.ACTIVE
|
||||
]
|
||||
for k in stale:
|
||||
del nomadnet_cached_links[k]
|
||||
to_teardown.append(nomadnet_cached_links.pop(k, None))
|
||||
_nomadnet_link_last_used.pop(k, None)
|
||||
|
||||
idle = [
|
||||
k
|
||||
for k, last in _nomadnet_link_last_used.items()
|
||||
if k in nomadnet_cached_links and now - last > LINK_IDLE_TTL_S
|
||||
]
|
||||
for k in idle:
|
||||
to_teardown.append(nomadnet_cached_links.pop(k, None))
|
||||
_nomadnet_link_last_used.pop(k, None)
|
||||
|
||||
while len(nomadnet_cached_links) > MAX_CACHED_LINKS:
|
||||
oldest_key = min(
|
||||
nomadnet_cached_links.keys(),
|
||||
key=lambda k: _nomadnet_link_last_used.get(k, 0.0),
|
||||
)
|
||||
to_teardown.append(nomadnet_cached_links.pop(oldest_key, None))
|
||||
_nomadnet_link_last_used.pop(oldest_key, None)
|
||||
|
||||
orphans = [k for k in _nomadnet_link_last_used if k not in nomadnet_cached_links]
|
||||
for k in orphans:
|
||||
del _nomadnet_link_last_used[k]
|
||||
_teardown_links(to_teardown)
|
||||
|
||||
|
||||
def _cache_link_if_active(destination_hash: bytes, link) -> None:
|
||||
if link is None or link.status is not RNS.Link.ACTIVE:
|
||||
return
|
||||
to_teardown = []
|
||||
with _nomadnet_links_lock:
|
||||
nomadnet_cached_links[destination_hash] = link
|
||||
_nomadnet_link_last_used[destination_hash] = time.time()
|
||||
while len(nomadnet_cached_links) > MAX_CACHED_LINKS:
|
||||
candidates = [
|
||||
k for k in nomadnet_cached_links if k != destination_hash
|
||||
]
|
||||
if not candidates:
|
||||
break
|
||||
oldest_key = min(
|
||||
candidates,
|
||||
key=lambda k: _nomadnet_link_last_used.get(k, 0.0),
|
||||
)
|
||||
to_teardown.append(nomadnet_cached_links.pop(oldest_key, None))
|
||||
_nomadnet_link_last_used.pop(oldest_key, None)
|
||||
_teardown_links(to_teardown)
|
||||
|
||||
|
||||
def _uncache_link_if_matches(destination_hash: bytes, link) -> None:
|
||||
|
|
@ -64,6 +126,7 @@ def _uncache_link_if_matches(destination_hash: bytes, link) -> None:
|
|||
del nomadnet_cached_links[destination_hash]
|
||||
except KeyError:
|
||||
pass
|
||||
_nomadnet_link_last_used.pop(destination_hash, None)
|
||||
|
||||
|
||||
class NomadnetDownloader:
|
||||
|
|
|
|||
|
|
@ -29,6 +29,7 @@ class HealthMonitor:
|
|||
ENTROPY_WARN_THRESHOLD = 1.5 # out of max ~2.32 for 5 log levels
|
||||
ERROR_RATE_WARN = 0.3
|
||||
MEMORY_WARN_MB = 100 # warn when available < 100 MB
|
||||
MEMORY_RECOVER_MB = 400 # restore SQLite RAM pragmas above this
|
||||
CONSECUTIVE_NEEDED = 2 # consecutive bad readings before alert
|
||||
|
||||
def __init__(self, log_handler, app=None):
|
||||
|
|
@ -41,6 +42,7 @@ class HealthMonitor:
|
|||
self._entropy_history = collections.deque(maxlen=self.ENTROPY_WINDOW)
|
||||
self._error_rate_history = collections.deque(maxlen=self.ENTROPY_WINDOW)
|
||||
self._mem_available_history = collections.deque(maxlen=self.ENTROPY_WINDOW)
|
||||
self._memory_pressure_active = False
|
||||
|
||||
def start(self):
|
||||
if self._running:
|
||||
|
|
@ -123,6 +125,15 @@ class HealthMonitor:
|
|||
"value": round(available_mb, 1),
|
||||
},
|
||||
)
|
||||
self._trigger_memory_pressure(available_mb)
|
||||
elif (
|
||||
self._memory_pressure_active
|
||||
and self._consecutive_above(
|
||||
self._mem_available_history,
|
||||
self.MEMORY_RECOVER_MB,
|
||||
)
|
||||
):
|
||||
self._recover_memory_pressure()
|
||||
|
||||
for w in warnings:
|
||||
_log.warning("Health warning: %s", w["message"])
|
||||
|
|
@ -153,6 +164,26 @@ class HealthMonitor:
|
|||
vals = list(deq)
|
||||
return vals[-1] < threshold and vals[-2] < threshold
|
||||
|
||||
def _trigger_memory_pressure(self, available_mb: float) -> None:
|
||||
self._memory_pressure_active = True
|
||||
manager = getattr(self.app, "memory_pressure", None) if self.app else None
|
||||
if manager is None:
|
||||
return
|
||||
try:
|
||||
manager.on_memory_low(available_mb)
|
||||
except Exception as exc:
|
||||
_log.debug("Memory pressure cleanup failed: %s", exc)
|
||||
|
||||
def _recover_memory_pressure(self) -> None:
|
||||
self._memory_pressure_active = False
|
||||
manager = getattr(self.app, "memory_pressure", None) if self.app else None
|
||||
if manager is None:
|
||||
return
|
||||
try:
|
||||
manager.on_memory_recovered()
|
||||
except Exception as exc:
|
||||
_log.debug("Memory pressure recovery failed: %s", exc)
|
||||
|
||||
def _broadcast(self, warning_data):
|
||||
if not self.app:
|
||||
return
|
||||
|
|
|
|||
|
|
@ -29,6 +29,11 @@ def format_outbound_path_finding_measure(outcome: OutboundPathOutcome) -> str:
|
|||
IDX_PT_TIMESTAMP = 0
|
||||
IDX_PT_RVCD_IF = 5
|
||||
|
||||
# Soft cap for in-memory RNS path table entries (~270 MB at RNS estimate).
|
||||
DEFAULT_PATH_TABLE_SOFT_CAP = 15_000
|
||||
DEFAULT_PATH_PRUNE_BATCH = 500
|
||||
DEFAULT_PATH_PRUNE_TRIGGER = 10_000
|
||||
|
||||
|
||||
def _path_table_entry_is_expired_by_reticulum_rules(entry) -> bool:
|
||||
ts = entry[IDX_PT_TIMESTAMP]
|
||||
|
|
@ -54,6 +59,126 @@ def transport_path_table_entry_is_expired(destination_hash: bytes) -> bool:
|
|||
return _path_table_entry_is_expired_by_reticulum_rules(entry)
|
||||
|
||||
|
||||
def path_table_size() -> int:
|
||||
try:
|
||||
with RNS.Transport.path_table_lock:
|
||||
return len(RNS.Transport.path_table)
|
||||
except Exception:
|
||||
return 0
|
||||
|
||||
|
||||
def prune_expired_path_table_entries(
|
||||
reticulum: Optional["ReticulumLike"] = None,
|
||||
*,
|
||||
max_to_drop: int = DEFAULT_PATH_PRUNE_BATCH,
|
||||
trigger_size: int = DEFAULT_PATH_PRUNE_TRIGGER,
|
||||
) -> int:
|
||||
"""Drop expired path-table entries when the table is large.
|
||||
|
||||
Returns the number of paths dropped.
|
||||
"""
|
||||
if max_to_drop <= 0:
|
||||
return 0
|
||||
size = path_table_size()
|
||||
if size < trigger_size:
|
||||
return 0
|
||||
|
||||
expired: list[tuple[float, bytes]] = []
|
||||
try:
|
||||
with RNS.Transport.path_table_lock:
|
||||
for dest_hash, entry in RNS.Transport.path_table.items():
|
||||
if _path_table_entry_is_expired_by_reticulum_rules(entry):
|
||||
ts = entry[IDX_PT_TIMESTAMP] if entry else 0.0
|
||||
expired.append((ts, dest_hash))
|
||||
except Exception:
|
||||
return 0
|
||||
|
||||
if not expired:
|
||||
return 0
|
||||
|
||||
expired.sort(key=lambda item: item[0])
|
||||
dropped = 0
|
||||
for _ts, dest_hash in expired[:max_to_drop]:
|
||||
try:
|
||||
if reticulum is not None:
|
||||
if reticulum.drop_path(dest_hash):
|
||||
dropped += 1
|
||||
else:
|
||||
RNS.Transport.expire_path(dest_hash)
|
||||
dropped += 1
|
||||
except Exception:
|
||||
continue
|
||||
return dropped
|
||||
|
||||
|
||||
def prune_path_table_to_soft_cap(
|
||||
reticulum: Optional["ReticulumLike"] = None,
|
||||
*,
|
||||
soft_cap: int = DEFAULT_PATH_TABLE_SOFT_CAP,
|
||||
max_to_drop: int = DEFAULT_PATH_PRUNE_BATCH,
|
||||
) -> int:
|
||||
"""If still over soft_cap after expiry prune, drop oldest unresponsive paths."""
|
||||
if soft_cap <= 0 or max_to_drop <= 0:
|
||||
return 0
|
||||
size = path_table_size()
|
||||
if size <= soft_cap:
|
||||
return 0
|
||||
|
||||
candidates: list[tuple[float, bytes]] = []
|
||||
try:
|
||||
with RNS.Transport.path_table_lock:
|
||||
for dest_hash, entry in RNS.Transport.path_table.items():
|
||||
ts = entry[IDX_PT_TIMESTAMP] if entry else 0.0
|
||||
unresponsive = False
|
||||
try:
|
||||
unresponsive = RNS.Transport.path_is_unresponsive(dest_hash)
|
||||
except Exception:
|
||||
unresponsive = False
|
||||
# Prefer unresponsive, then oldest.
|
||||
rank = (0.0 if unresponsive else 1.0e12) + float(ts or 0.0)
|
||||
candidates.append((rank, dest_hash))
|
||||
except Exception:
|
||||
return 0
|
||||
|
||||
if not candidates:
|
||||
return 0
|
||||
|
||||
over = size - soft_cap
|
||||
budget = min(max_to_drop, over)
|
||||
candidates.sort(key=lambda item: item[0])
|
||||
dropped = 0
|
||||
for _rank, dest_hash in candidates[:budget]:
|
||||
try:
|
||||
if reticulum is not None:
|
||||
if reticulum.drop_path(dest_hash):
|
||||
dropped += 1
|
||||
else:
|
||||
RNS.Transport.expire_path(dest_hash)
|
||||
dropped += 1
|
||||
except Exception:
|
||||
continue
|
||||
return dropped
|
||||
|
||||
|
||||
def clean_rns_announce_cache() -> bool:
|
||||
"""Ask RNS to trim on-disk announce cache files not referenced by paths."""
|
||||
try:
|
||||
owner = getattr(RNS.Transport, "owner", None)
|
||||
if owner is not None and getattr(owner, "is_connected_to_shared_instance", False):
|
||||
return False
|
||||
clean = getattr(RNS.Transport, "clean_cache", None)
|
||||
if callable(clean):
|
||||
clean()
|
||||
return True
|
||||
clean_ann = getattr(RNS.Transport, "clean_announce_cache", None)
|
||||
if callable(clean_ann):
|
||||
clean_ann()
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
return False
|
||||
|
||||
|
||||
def should_rediscover_path(destination_hash: bytes) -> bool:
|
||||
if not RNS.Transport.has_path(destination_hash):
|
||||
return True
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ from meshchatx.src.backend import reticulum_pathfinding
|
|||
# Kept separate from nomadnet_downloader.nomadnet_cached_links — the two caches
|
||||
# may merge in the future if NomadNet is ported onto this generic Links API.
|
||||
rns_cached_links: dict[tuple[str, bytes], "RNS.Link"] = {}
|
||||
_rns_link_last_used: dict[tuple[str, bytes], float] = {}
|
||||
_rns_links_lock = threading.Lock()
|
||||
|
||||
# Per-cache-key count of consecutive RNS.Link request failures. Reset on
|
||||
|
|
@ -29,10 +30,19 @@ _link_failure_counts: dict[tuple[str, bytes], int] = {}
|
|||
# then go through the full open_link path and re-establish.
|
||||
_LINK_RECYCLE_FAILURE_THRESHOLD = 2
|
||||
|
||||
# Cap active RNS links retained in process memory.
|
||||
MAX_CACHED_LINKS = 64
|
||||
LINK_IDLE_TTL_S = 30 * 60
|
||||
|
||||
# Wait granularity while polling for path / link (seconds).
|
||||
_POLL_INTERVAL_S = 0.02
|
||||
|
||||
|
||||
def cached_link_count() -> int:
|
||||
with _rns_links_lock:
|
||||
return len(rns_cached_links)
|
||||
|
||||
|
||||
def get_cached_active_link(aspect: str, destination_hash: bytes):
|
||||
"""Return a cached link if present and ACTIVE; drop stale entries."""
|
||||
key = (aspect, destination_hash)
|
||||
|
|
@ -41,37 +51,94 @@ def get_cached_active_link(aspect: str, destination_hash: bytes):
|
|||
if link is None:
|
||||
return None
|
||||
if link.status is RNS.Link.ACTIVE:
|
||||
_rns_link_last_used[key] = time.time()
|
||||
return link
|
||||
try:
|
||||
del rns_cached_links[key]
|
||||
except KeyError:
|
||||
pass
|
||||
_rns_link_last_used.pop(key, None)
|
||||
_link_failure_counts.pop(key, None)
|
||||
return None
|
||||
|
||||
|
||||
def _teardown_links(links) -> None:
|
||||
for link in links:
|
||||
if link is None:
|
||||
continue
|
||||
try:
|
||||
link.teardown()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def _evict_over_cap_locked(preserve_key=None):
|
||||
"""Evict oldest links until at or under MAX_CACHED_LINKS. Caller holds lock."""
|
||||
to_teardown = []
|
||||
while len(rns_cached_links) > MAX_CACHED_LINKS:
|
||||
candidates = [
|
||||
k for k in rns_cached_links if preserve_key is None or k != preserve_key
|
||||
]
|
||||
if not candidates:
|
||||
break
|
||||
oldest_key = min(
|
||||
candidates,
|
||||
key=lambda k: _rns_link_last_used.get(k, 0.0),
|
||||
)
|
||||
to_teardown.append(rns_cached_links.pop(oldest_key, None))
|
||||
_rns_link_last_used.pop(oldest_key, None)
|
||||
_link_failure_counts.pop(oldest_key, None)
|
||||
return to_teardown
|
||||
|
||||
|
||||
def sweep_stale_links():
|
||||
now = time.time()
|
||||
to_teardown = []
|
||||
with _rns_links_lock:
|
||||
stale = [
|
||||
k for k, v in rns_cached_links.items() if v.status is not RNS.Link.ACTIVE
|
||||
]
|
||||
for k in stale:
|
||||
del rns_cached_links[k]
|
||||
to_teardown.append(rns_cached_links.pop(k, None))
|
||||
_rns_link_last_used.pop(k, None)
|
||||
_link_failure_counts.pop(k, None)
|
||||
|
||||
idle = [
|
||||
k
|
||||
for k, last in _rns_link_last_used.items()
|
||||
if k in rns_cached_links and now - last > LINK_IDLE_TTL_S
|
||||
]
|
||||
for k in idle:
|
||||
to_teardown.append(rns_cached_links.pop(k, None))
|
||||
_rns_link_last_used.pop(k, None)
|
||||
_link_failure_counts.pop(k, None)
|
||||
|
||||
to_teardown.extend(_evict_over_cap_locked())
|
||||
|
||||
# Drop counter entries whose link is no longer cached so the dict
|
||||
# cannot grow unbounded across link churn.
|
||||
orphans = [k for k in _link_failure_counts if k not in rns_cached_links]
|
||||
for k in orphans:
|
||||
del _link_failure_counts[k]
|
||||
used_orphans = [k for k in _rns_link_last_used if k not in rns_cached_links]
|
||||
for k in used_orphans:
|
||||
del _rns_link_last_used[k]
|
||||
_teardown_links(to_teardown)
|
||||
|
||||
|
||||
def _cache_link_if_active(aspect: str, destination_hash: bytes, link) -> None:
|
||||
if link is None or link.status is not RNS.Link.ACTIVE:
|
||||
return
|
||||
key = (aspect, destination_hash)
|
||||
to_teardown = []
|
||||
with _rns_links_lock:
|
||||
rns_cached_links[key] = link
|
||||
_rns_link_last_used[key] = time.time()
|
||||
# A freshly cached link starts with a clean failure count, even if
|
||||
# an older link at the same key died with a non-zero count.
|
||||
_link_failure_counts.pop(key, None)
|
||||
to_teardown.extend(_evict_over_cap_locked(preserve_key=key))
|
||||
_teardown_links(to_teardown)
|
||||
|
||||
|
||||
def _uncache_link_if_matches(aspect: str, destination_hash: bytes, link) -> None:
|
||||
|
|
@ -84,6 +151,7 @@ def _uncache_link_if_matches(aspect: str, destination_hash: bytes, link) -> None
|
|||
del rns_cached_links[key]
|
||||
except KeyError:
|
||||
pass
|
||||
_rns_link_last_used.pop(key, None)
|
||||
_link_failure_counts.pop(key, None)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1911,6 +1911,7 @@ export default {
|
|||
audioAttachmentRecordingTimer: null,
|
||||
androidNativeOpusAttachment: false,
|
||||
lxmfMessageAudioAttachmentCache: {},
|
||||
lxmfMessageAudioAttachmentOrder: [],
|
||||
isDownloadingAudio: {},
|
||||
expandedMessageInfo: null,
|
||||
imageModalUrl: null,
|
||||
|
|
@ -2418,6 +2419,7 @@ export default {
|
|||
this.messageBubbleTranslation = {};
|
||||
if (oldPeer) {
|
||||
this.saveDraft(oldPeer.destination_hash);
|
||||
this.clearAudioAttachmentCache();
|
||||
}
|
||||
this.teardownPeerHeaderResizeObserver();
|
||||
this.disconnectOpenConversationScrollObserver();
|
||||
|
|
@ -2547,8 +2549,56 @@ export default {
|
|||
clearInterval(this.propagationStatusInterval);
|
||||
}
|
||||
this.disconnectOpenConversationScrollObserver();
|
||||
this.clearAudioAttachmentCache();
|
||||
},
|
||||
methods: {
|
||||
clearAudioAttachmentCache() {
|
||||
const cache = this.lxmfMessageAudioAttachmentCache || {};
|
||||
for (const url of Object.values(cache)) {
|
||||
if (typeof url === "string" && url.startsWith("blob:")) {
|
||||
try {
|
||||
URL.revokeObjectURL(url);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
}
|
||||
}
|
||||
this.lxmfMessageAudioAttachmentCache = {};
|
||||
this.lxmfMessageAudioAttachmentOrder = [];
|
||||
},
|
||||
rememberAudioAttachment(hash, objectUrl) {
|
||||
if (!hash || !objectUrl) {
|
||||
return;
|
||||
}
|
||||
const maxEntries = 50;
|
||||
const cache = this.lxmfMessageAudioAttachmentCache;
|
||||
const order = this.lxmfMessageAudioAttachmentOrder;
|
||||
if (cache[hash] && cache[hash] !== objectUrl) {
|
||||
try {
|
||||
URL.revokeObjectURL(cache[hash]);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
}
|
||||
cache[hash] = objectUrl;
|
||||
const existing = order.indexOf(hash);
|
||||
if (existing >= 0) {
|
||||
order.splice(existing, 1);
|
||||
}
|
||||
order.push(hash);
|
||||
while (order.length > maxEntries) {
|
||||
const evictHash = order.shift();
|
||||
const evictUrl = cache[evictHash];
|
||||
delete cache[evictHash];
|
||||
if (typeof evictUrl === "string" && evictUrl.startsWith("blob:")) {
|
||||
try {
|
||||
URL.revokeObjectURL(evictUrl);
|
||||
} catch {
|
||||
/* ignore */
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
isMeshChatXAndroid() {
|
||||
return (
|
||||
window.MeshChatXAndroid &&
|
||||
|
|
@ -4281,7 +4331,7 @@ export default {
|
|||
|
||||
const objectUrl = await this.decodeLxmfAudioFieldToBlobUrl(audioField);
|
||||
if (objectUrl) {
|
||||
this.lxmfMessageAudioAttachmentCache[chatItem.lxmf_message.hash] = objectUrl;
|
||||
this.rememberAudioAttachment(chatItem.lxmf_message.hash, objectUrl);
|
||||
}
|
||||
} catch (e) {
|
||||
console.error("Failed to download or decode audio:", e);
|
||||
|
|
@ -5142,8 +5192,7 @@ export default {
|
|||
continue;
|
||||
}
|
||||
|
||||
// update audio cache
|
||||
this.lxmfMessageAudioAttachmentCache[chatItem.lxmf_message.hash] = objectUrl;
|
||||
this.rememberAudioAttachment(chatItem.lxmf_message.hash, objectUrl);
|
||||
}
|
||||
},
|
||||
async decodeLxmfAudioFieldToBlobUrl(audioField) {
|
||||
|
|
|
|||
|
|
@ -53,6 +53,8 @@ import NetworkVisualiserLegend from "./internal/NetworkVisualiserLegend.vue";
|
|||
import {
|
||||
ANNOUNCE_HASH_CHUNK_SIZE,
|
||||
VIZ_ANNOUNCE_ASPECTS,
|
||||
VIZ_ANNOUNCE_SOFT_CAP,
|
||||
VIZ_PATH_TABLE_SOFT_CAP,
|
||||
dedupeIconQueueEntries,
|
||||
pathHashesWithinHopFilter,
|
||||
pickAdaptiveFetchConcurrency,
|
||||
|
|
@ -259,6 +261,15 @@ export default {
|
|||
delete this.iconCache[key];
|
||||
}
|
||||
this.iconCache = {};
|
||||
this.pathTable = [];
|
||||
this.announces = {};
|
||||
this.conversations = {};
|
||||
try {
|
||||
this.nodes.clear();
|
||||
this.edges.clear();
|
||||
} catch {
|
||||
/* DataSet may already be destroyed with the network */
|
||||
}
|
||||
},
|
||||
mounted() {
|
||||
const isMobile = window.innerWidth < 640;
|
||||
|
|
@ -324,10 +335,15 @@ export default {
|
|||
});
|
||||
this.pathTable.push(...firstResp.data.path_table);
|
||||
const totalCount = firstResp.data.total_count;
|
||||
const softCap = VIZ_PATH_TABLE_SOFT_CAP;
|
||||
if (totalCount > this.pageSize) {
|
||||
const concurrency = this.pathFetchConcurrency;
|
||||
for (let offset = this.pageSize; offset < totalCount; offset += this.pageSize * concurrency) {
|
||||
if (this.abortController.signal.aborted) return;
|
||||
if (this.pathTable.length >= softCap) {
|
||||
this.loadingStatus = `Loading paths (capped at ${softCap} / ${totalCount})`;
|
||||
break;
|
||||
}
|
||||
const chunk = [];
|
||||
for (let i = 0; i < concurrency && offset + i * this.pageSize < totalCount; i++) {
|
||||
chunk.push(offset + i * this.pageSize);
|
||||
|
|
@ -340,7 +356,10 @@ export default {
|
|||
);
|
||||
const responses = await Promise.all(promises);
|
||||
for (const r of responses) {
|
||||
this.pathTable.push(...r.data.path_table);
|
||||
const rows = r.data.path_table || [];
|
||||
const room = softCap - this.pathTable.length;
|
||||
if (room <= 0) break;
|
||||
this.pathTable.push(...rows.slice(0, room));
|
||||
}
|
||||
this.loadingStatus = `Loading paths (${this.pathTable.length} / ${totalCount})`;
|
||||
}
|
||||
|
|
@ -402,6 +421,15 @@ export default {
|
|||
}
|
||||
}
|
||||
}
|
||||
const announceKeys = Object.keys(this.announces);
|
||||
if (announceKeys.length > VIZ_ANNOUNCE_SOFT_CAP) {
|
||||
const neededSet = new Set(needed);
|
||||
const extras = announceKeys.filter((hash) => !neededSet.has(hash));
|
||||
const overflow = announceKeys.length - VIZ_ANNOUNCE_SOFT_CAP;
|
||||
for (const hash of extras.slice(0, overflow)) {
|
||||
delete this.announces[hash];
|
||||
}
|
||||
}
|
||||
},
|
||||
async getConfig() {
|
||||
try {
|
||||
|
|
|
|||
|
|
@ -1,7 +1,12 @@
|
|||
const DB_NAME = "meshchat_map_cache";
|
||||
const DB_VERSION = 2;
|
||||
const DB_VERSION = 3;
|
||||
const STORE_NAME = "tiles";
|
||||
const STATE_STORE = "map_state";
|
||||
const META_STORE = "tile_meta";
|
||||
|
||||
const MAX_TILES = 5000;
|
||||
const MAX_BYTES = 256 * 1024 * 1024;
|
||||
const TILE_TTL_MS = 30 * 24 * 60 * 60 * 1000;
|
||||
|
||||
class TileCache {
|
||||
constructor() {
|
||||
|
|
@ -36,6 +41,9 @@ class TileCache {
|
|||
if (!db.objectStoreNames.contains(STATE_STORE)) {
|
||||
db.createObjectStore(STATE_STORE);
|
||||
}
|
||||
if (!db.objectStoreNames.contains(META_STORE)) {
|
||||
db.createObjectStore(META_STORE);
|
||||
}
|
||||
};
|
||||
|
||||
request.onsuccess = (event) => {
|
||||
|
|
@ -45,24 +53,59 @@ class TileCache {
|
|||
});
|
||||
}
|
||||
|
||||
_blobSize(data) {
|
||||
if (!data) return 0;
|
||||
if (typeof data.size === "number") return data.size;
|
||||
if (data.byteLength != null) return data.byteLength;
|
||||
if (typeof data === "string") return data.length;
|
||||
return 0;
|
||||
}
|
||||
|
||||
async getTile(key) {
|
||||
await this.initPromise;
|
||||
return new Promise((resolve, reject) => {
|
||||
const transaction = this.db.transaction([STORE_NAME], "readonly");
|
||||
const transaction = this.db.transaction([STORE_NAME, META_STORE], "readwrite");
|
||||
const store = transaction.objectStore(STORE_NAME);
|
||||
const metaStore = transaction.objectStore(META_STORE);
|
||||
const request = store.get(key);
|
||||
|
||||
request.onsuccess = () => resolve(request.result);
|
||||
request.onsuccess = () => {
|
||||
const value = request.result;
|
||||
if (value != null) {
|
||||
const metaReq = metaStore.get(key);
|
||||
metaReq.onsuccess = () => {
|
||||
const prev = metaReq.result || {};
|
||||
metaStore.put(
|
||||
{
|
||||
...prev,
|
||||
lastAccess: Date.now(),
|
||||
size: prev.size ?? this._blobSize(value),
|
||||
},
|
||||
key
|
||||
);
|
||||
};
|
||||
}
|
||||
resolve(value);
|
||||
};
|
||||
request.onerror = () => reject(request.error);
|
||||
});
|
||||
}
|
||||
|
||||
async setTile(key, data) {
|
||||
await this.initPromise;
|
||||
await this._evictIfNeeded(this._blobSize(data));
|
||||
return new Promise((resolve, reject) => {
|
||||
const transaction = this.db.transaction([STORE_NAME], "readwrite");
|
||||
const transaction = this.db.transaction([STORE_NAME, META_STORE], "readwrite");
|
||||
const store = transaction.objectStore(STORE_NAME);
|
||||
const metaStore = transaction.objectStore(META_STORE);
|
||||
store.put(data, key);
|
||||
metaStore.put(
|
||||
{
|
||||
lastAccess: Date.now(),
|
||||
size: this._blobSize(data),
|
||||
},
|
||||
key
|
||||
);
|
||||
|
||||
transaction.oncomplete = () => resolve();
|
||||
transaction.onerror = () => reject(transaction.error);
|
||||
|
|
@ -70,6 +113,76 @@ class TileCache {
|
|||
});
|
||||
}
|
||||
|
||||
async _readAllMeta() {
|
||||
return new Promise((resolve, reject) => {
|
||||
const transaction = this.db.transaction([META_STORE], "readonly");
|
||||
const store = transaction.objectStore(META_STORE);
|
||||
const request = store.openCursor();
|
||||
const rows = [];
|
||||
request.onsuccess = (event) => {
|
||||
const cursor = event.target.result;
|
||||
if (!cursor) {
|
||||
resolve(rows);
|
||||
return;
|
||||
}
|
||||
rows.push({ key: cursor.key, ...(cursor.value || {}) });
|
||||
cursor.continue();
|
||||
};
|
||||
request.onerror = () => reject(request.error);
|
||||
});
|
||||
}
|
||||
|
||||
async _evictIfNeeded(incomingBytes = 0) {
|
||||
let meta;
|
||||
try {
|
||||
meta = await this._readAllMeta();
|
||||
} catch {
|
||||
return;
|
||||
}
|
||||
const now = Date.now();
|
||||
const expired = meta.filter((m) => m.lastAccess && now - m.lastAccess > TILE_TTL_MS);
|
||||
if (expired.length > 0) {
|
||||
await this._deleteKeys(expired.map((m) => m.key));
|
||||
meta = meta.filter((m) => !expired.some((e) => e.key === m.key));
|
||||
}
|
||||
|
||||
let totalBytes = meta.reduce((sum, m) => sum + (m.size || 0), 0) + incomingBytes;
|
||||
let count = meta.length + (incomingBytes > 0 ? 1 : 0);
|
||||
if (count <= MAX_TILES && totalBytes <= MAX_BYTES) {
|
||||
return;
|
||||
}
|
||||
|
||||
const ordered = meta.slice().sort((a, b) => (a.lastAccess || 0) - (b.lastAccess || 0));
|
||||
const toDelete = [];
|
||||
for (const row of ordered) {
|
||||
if (count <= MAX_TILES && totalBytes <= MAX_BYTES) {
|
||||
break;
|
||||
}
|
||||
toDelete.push(row.key);
|
||||
totalBytes -= row.size || 0;
|
||||
count -= 1;
|
||||
}
|
||||
if (toDelete.length > 0) {
|
||||
await this._deleteKeys(toDelete);
|
||||
}
|
||||
}
|
||||
|
||||
async _deleteKeys(keys) {
|
||||
if (!keys.length) return;
|
||||
return new Promise((resolve, reject) => {
|
||||
const transaction = this.db.transaction([STORE_NAME, META_STORE], "readwrite");
|
||||
const store = transaction.objectStore(STORE_NAME);
|
||||
const metaStore = transaction.objectStore(META_STORE);
|
||||
for (const key of keys) {
|
||||
store.delete(key);
|
||||
metaStore.delete(key);
|
||||
}
|
||||
transaction.oncomplete = () => resolve();
|
||||
transaction.onerror = () => reject(transaction.error);
|
||||
transaction.onabort = () => reject(transaction.error || new Error("IndexedDB transaction aborted"));
|
||||
});
|
||||
}
|
||||
|
||||
async getMapState(key) {
|
||||
await this.initPromise;
|
||||
return new Promise((resolve, reject) => {
|
||||
|
|
@ -98,9 +211,14 @@ class TileCache {
|
|||
async clear() {
|
||||
await this.initPromise;
|
||||
return new Promise((resolve, reject) => {
|
||||
const transaction = this.db.transaction([STORE_NAME, STATE_STORE], "readwrite");
|
||||
transaction.objectStore(STORE_NAME).clear();
|
||||
transaction.objectStore(STATE_STORE).clear();
|
||||
const stores = [STORE_NAME, STATE_STORE];
|
||||
if (this.db.objectStoreNames.contains(META_STORE)) {
|
||||
stores.push(META_STORE);
|
||||
}
|
||||
const transaction = this.db.transaction(stores, "readwrite");
|
||||
for (const name of stores) {
|
||||
transaction.objectStore(name).clear();
|
||||
}
|
||||
|
||||
transaction.oncomplete = () => resolve();
|
||||
transaction.onerror = () => reject(transaction.error);
|
||||
|
|
|
|||
|
|
@ -4,6 +4,12 @@ export const VIZ_ANNOUNCE_ASPECTS = ["lxmf.delivery", "nomadnetwork.node"];
|
|||
|
||||
export const ANNOUNCE_HASH_CHUNK_SIZE = 500;
|
||||
|
||||
/** Soft cap for client-side path table rows kept in the visualiser. */
|
||||
export const VIZ_PATH_TABLE_SOFT_CAP = 20_000;
|
||||
|
||||
/** Soft cap for announce map entries keyed by destination hash. */
|
||||
export const VIZ_ANNOUNCE_SOFT_CAP = 10_000;
|
||||
|
||||
/**
|
||||
* @param {unknown[]} pathTable
|
||||
* @param {number|null|undefined} hopMax
|
||||
|
|
|
|||
69
tests/backend/test_memory_pressure.py
Normal file
69
tests/backend/test_memory_pressure.py
Normal file
|
|
@ -0,0 +1,69 @@
|
|||
# SPDX-License-Identifier: 0BSD
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from meshchatx.src.backend.memory_pressure import MemoryPressureManager
|
||||
|
||||
|
||||
def test_run_periodic_cleanup_sweeps_and_reports_stats():
|
||||
app = MagicMock()
|
||||
app.reticulum = MagicMock()
|
||||
app.rnpath_handler = MagicMock()
|
||||
app.database = MagicMock()
|
||||
manager = MemoryPressureManager(app=app)
|
||||
|
||||
with (
|
||||
patch(
|
||||
"meshchatx.src.backend.memory_pressure.nomadnet_downloader.cached_link_count",
|
||||
side_effect=[2, 1],
|
||||
),
|
||||
patch(
|
||||
"meshchatx.src.backend.memory_pressure.nomadnet_downloader.sweep_stale_links",
|
||||
) as nomad_sweep,
|
||||
patch(
|
||||
"meshchatx.src.backend.memory_pressure.rns_link_manager.cached_link_count",
|
||||
side_effect=[3, 2],
|
||||
),
|
||||
patch(
|
||||
"meshchatx.src.backend.memory_pressure.rns_link_manager.sweep_stale_links",
|
||||
) as rns_sweep,
|
||||
patch(
|
||||
"meshchatx.src.backend.memory_pressure.reticulum_pathfinding.prune_expired_path_table_entries",
|
||||
return_value=4,
|
||||
),
|
||||
patch(
|
||||
"meshchatx.src.backend.memory_pressure.reticulum_pathfinding.prune_path_table_to_soft_cap",
|
||||
return_value=1,
|
||||
),
|
||||
patch(
|
||||
"meshchatx.src.backend.memory_pressure.reticulum_pathfinding.clean_rns_announce_cache",
|
||||
return_value=True,
|
||||
),
|
||||
patch(
|
||||
"meshchatx.src.backend.memory_pressure.reticulum_pathfinding.path_table_size",
|
||||
return_value=42,
|
||||
),
|
||||
patch.object(manager, "_count_held_announces", return_value=0),
|
||||
):
|
||||
stats = manager.run_periodic_cleanup()
|
||||
|
||||
nomad_sweep.assert_called_once()
|
||||
rns_sweep.assert_called_once()
|
||||
assert stats["nomad_links_swept"] == 1
|
||||
assert stats["rns_links_swept"] == 1
|
||||
assert stats["paths_pruned_expired"] == 4
|
||||
assert stats["paths_pruned_cap"] == 1
|
||||
assert stats["announce_cache_cleaned"] is True
|
||||
assert stats["path_table_size"] == 42
|
||||
|
||||
|
||||
def test_on_memory_low_relaxes_sqlite():
|
||||
app = MagicMock()
|
||||
app.database = MagicMock()
|
||||
manager = MemoryPressureManager(app=app)
|
||||
with patch.object(manager, "run_periodic_cleanup", return_value={"ok": True}):
|
||||
stats = manager.on_memory_low(50.0)
|
||||
app.database.apply_memory_pressure_pragmas.assert_called_once_with(True)
|
||||
assert stats["sqlite_relaxed"] is True
|
||||
manager.on_memory_recovered()
|
||||
app.database.apply_memory_pressure_pragmas.assert_called_with(False)
|
||||
|
|
@ -19,9 +19,15 @@ from meshchatx.src.backend.nomadnet_downloader import (
|
|||
def clear_nomadnet_link_cache():
|
||||
with _nomadnet_links_lock:
|
||||
nomadnet_cached_links.clear()
|
||||
from meshchatx.src.backend.nomadnet_downloader import _nomadnet_link_last_used
|
||||
|
||||
_nomadnet_link_last_used.clear()
|
||||
yield
|
||||
with _nomadnet_links_lock:
|
||||
nomadnet_cached_links.clear()
|
||||
from meshchatx.src.backend.nomadnet_downloader import _nomadnet_link_last_used
|
||||
|
||||
_nomadnet_link_last_used.clear()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -159,3 +165,25 @@ def test_file_downloader_passes_query_data_to_parent():
|
|||
data="foo=bar",
|
||||
)
|
||||
assert fd.data == "foo=bar"
|
||||
|
||||
|
||||
def test_nomad_link_cache_evicts_over_cap():
|
||||
from meshchatx.src.backend import nomadnet_downloader as nd
|
||||
|
||||
original_max = nd.MAX_CACHED_LINKS
|
||||
nd.MAX_CACHED_LINKS = 2
|
||||
try:
|
||||
links = []
|
||||
for i in range(3):
|
||||
link = MagicMock()
|
||||
link.status = RNS.Link.ACTIVE
|
||||
dest = bytes([i]) * 16
|
||||
nd._cache_link_if_active(dest, link)
|
||||
links.append((dest, link))
|
||||
assert nd.cached_link_count() == 2
|
||||
assert get_cached_active_link(links[0][0]) is None
|
||||
links[0][1].teardown.assert_called()
|
||||
assert get_cached_active_link(links[1][0]) is links[1][1]
|
||||
assert get_cached_active_link(links[2][0]) is links[2][1]
|
||||
finally:
|
||||
nd.MAX_CACHED_LINKS = original_max
|
||||
|
|
|
|||
|
|
@ -356,3 +356,55 @@ async def test_wait_for_path_times_out():
|
|||
):
|
||||
ok = await rp.wait_for_path(MagicMock(), DEST, 0.01, 0.01)
|
||||
assert ok is False
|
||||
|
||||
|
||||
def test_prune_expired_path_table_entries_respects_trigger():
|
||||
cleanup = []
|
||||
try:
|
||||
old = time.time() - (RNS.Transport.DESTINATION_TIMEOUT + 100)
|
||||
entry = [old, None, None, None, None, None]
|
||||
_put_path_entry(DEST, entry, cleanup)
|
||||
assert (
|
||||
rp.prune_expired_path_table_entries(
|
||||
None,
|
||||
max_to_drop=10,
|
||||
trigger_size=10_000,
|
||||
)
|
||||
== 0
|
||||
)
|
||||
reticulum = MagicMock()
|
||||
reticulum.drop_path.return_value = True
|
||||
dropped = rp.prune_expired_path_table_entries(
|
||||
reticulum,
|
||||
max_to_drop=10,
|
||||
trigger_size=1,
|
||||
)
|
||||
assert dropped == 1
|
||||
reticulum.drop_path.assert_called_once_with(DEST)
|
||||
finally:
|
||||
with RNS.Transport.path_table_lock:
|
||||
for dest in cleanup:
|
||||
RNS.Transport.path_table.pop(dest, None)
|
||||
|
||||
|
||||
def test_prune_path_table_to_soft_cap_drops_oldest():
|
||||
cleanup = []
|
||||
try:
|
||||
reticulum = MagicMock()
|
||||
reticulum.drop_path.return_value = True
|
||||
for i in range(3):
|
||||
dest = bytes([i + 10]) * 16
|
||||
entry = [float(i), None, None, None, None, None]
|
||||
_put_path_entry(dest, entry, cleanup)
|
||||
with patch.object(RNS.Transport, "path_is_unresponsive", return_value=False):
|
||||
dropped = rp.prune_path_table_to_soft_cap(
|
||||
reticulum,
|
||||
soft_cap=1,
|
||||
max_to_drop=5,
|
||||
)
|
||||
assert dropped >= 1
|
||||
assert reticulum.drop_path.call_count >= 1
|
||||
finally:
|
||||
with RNS.Transport.path_table_lock:
|
||||
for dest in cleanup:
|
||||
RNS.Transport.path_table.pop(dest, None)
|
||||
|
|
|
|||
|
|
@ -14,10 +14,12 @@ from meshchatx.src.backend import rns_link_manager as rlm
|
|||
def clear_link_cache():
|
||||
with rlm._rns_links_lock:
|
||||
rlm.rns_cached_links.clear()
|
||||
rlm._rns_link_last_used.clear()
|
||||
rlm._link_failure_counts.clear()
|
||||
yield
|
||||
with rlm._rns_links_lock:
|
||||
rlm.rns_cached_links.clear()
|
||||
rlm._rns_link_last_used.clear()
|
||||
rlm._link_failure_counts.clear()
|
||||
|
||||
|
||||
|
|
@ -451,3 +453,23 @@ def test_parse_dest_aspect_helpers():
|
|||
{"destination_hash": "zz", "aspect": "a"}
|
||||
)
|
||||
assert err == "invalid_destination_hash"
|
||||
|
||||
|
||||
def test_rns_link_cache_evicts_over_cap():
|
||||
original_max = rlm.MAX_CACHED_LINKS
|
||||
rlm.MAX_CACHED_LINKS = 2
|
||||
try:
|
||||
kept = []
|
||||
for i in range(3):
|
||||
link = MagicMock()
|
||||
link.status = rlm.RNS.Link.ACTIVE
|
||||
dest = bytes([i + 1]) * 16
|
||||
rlm._cache_link_if_active("app.aspect", dest, link)
|
||||
kept.append((dest, link))
|
||||
assert rlm.cached_link_count() == 2
|
||||
assert rlm.get_cached_active_link("app.aspect", kept[0][0]) is None
|
||||
kept[0][1].teardown.assert_called()
|
||||
assert rlm.get_cached_active_link("app.aspect", kept[1][0]) is kept[1][1]
|
||||
assert rlm.get_cached_active_link("app.aspect", kept[2][0]) is kept[2][1]
|
||||
finally:
|
||||
rlm.MAX_CACHED_LINKS = original_max
|
||||
|
|
|
|||
14
tests/backend/test_sqlite_memory_pressure.py
Normal file
14
tests/backend/test_sqlite_memory_pressure.py
Normal file
|
|
@ -0,0 +1,14 @@
|
|||
# SPDX-License-Identifier: 0BSD
|
||||
|
||||
from meshchatx.src.backend.database import Database
|
||||
|
||||
|
||||
def test_apply_memory_pressure_pragmas_roundtrip(tmp_path):
|
||||
db = Database(str(tmp_path / "pressure.db"))
|
||||
db.initialize()
|
||||
assert db.apply_memory_pressure_pragmas(True) is True
|
||||
assert db._sqlite_memory_relaxed is True
|
||||
assert db._get_pragma_value("temp_store") == 1 # FILE
|
||||
assert db.apply_memory_pressure_pragmas(False) is True
|
||||
assert db._sqlite_memory_relaxed is False
|
||||
assert db._get_pragma_value("temp_store") == 2 # MEMORY
|
||||
26
tests/electron/backendProcessOrphans.test.js
Normal file
26
tests/electron/backendProcessOrphans.test.js
Normal file
|
|
@ -0,0 +1,26 @@
|
|||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
isHeadlessBackendArgs,
|
||||
} from "../../electron/backendProcessOrphans.js";
|
||||
|
||||
describe("electron/backendProcessOrphans", () => {
|
||||
it("detects headless backend command lines", () => {
|
||||
expect(
|
||||
isHeadlessBackendArgs(
|
||||
"/tmp/.mount_x/resources/backend/ReticulumMeshChatX --headless --port 9337"
|
||||
)
|
||||
).toBe(true);
|
||||
expect(
|
||||
isHeadlessBackendArgs(
|
||||
"/home/user/ReticulumMeshChatX --headless"
|
||||
)
|
||||
).toBe(true);
|
||||
});
|
||||
|
||||
it("ignores non-backend and non-headless processes", () => {
|
||||
expect(isHeadlessBackendArgs("reticulum-meshchatx --type=renderer")).toBe(false);
|
||||
expect(isHeadlessBackendArgs("/tmp/ReticulumMeshChatX")).toBe(false);
|
||||
expect(isHeadlessBackendArgs("")).toBe(false);
|
||||
expect(isHeadlessBackendArgs(null)).toBe(false);
|
||||
});
|
||||
});
|
||||
Loading…
Add table
Add a link
Reference in a new issue