Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/reticulum-sidecar-ipc.md
Original file line number Diff line number Diff line change
Expand Up @@ -246,7 +246,7 @@ Event types: `lxmf_message`, `lxmf_outbound_status`, `events_lagged` (WS subscri
- **`rrc.disconnected`:** payload `{ hub_dest_hash, reason, will_reconnect? }`. When `will_reconnect` is `false` (or `reason` is `local_disconnect`), the renderer drops that hub session. When `true` (or omitted on older sidecars), the UI shows reconnecting and keeps volatile rooms until WELCOME.

- **`lxmf_outbound_status`:** authoritative outbound delivery updates. Payload: `{ message_hash, status, delivery_method?, to_hash?, sent_via? }` where `status` is `delivered`, `failed`, or intermediate `sending` (egress upgrade or Direct→PN fallback). mesh-client maps `delivered` → UI Completes (`acked`) and persists `delivery_status` (+ `delivery_method` when present) to SQLite; Propagated Completes show **Stored at propagation node**; `failed` → Failed. Do **not** treat `/api/v1/lxmf/send` response `delivery_status` (`queued`/`sending`) as terminal. After Direct failure with a preferred remote PN, the sidecar re-queues once as Propagated and emits `sending` + `delivery_method: "propagated"` before a final `delivered`/`failed`.
- **`announce.received`:** coalesced WS notify for LXMF identity announces / path responses (named or nameless). Sidecar applies identity-key + display-name cache updates immediately, but emits **at most one** WS frame per coalesce window (500ms normal / 1000ms when >256 distinct destinations are pending) so announce storms stay O(1) bus pressure on large meshes (~100k). Payload is either a single `{ destination_hash, display_name?, hops }` (legacy / one-row flush) or `{ announces: [{ destination_hash, display_name?, hops }, ...] }` (capped at 1024, named preferred; overflow dropped — slow peer poll recovers). Each flush publishes pressure counters under `GET /api/v1/diagnostics` → `announce_ws` (ingress/unique/overflow + storm/flush timestamps) for the Diagnostics `reticulum/announce-bus-pressure` warning. Display names update the peer-label cache only — announces do **not** auto-create LXMF contacts. That cache is overlayed onto `GET /api/v1/peers` / topology rows **and** onto nameless/hash-prefix rows from `GET /api/v1/contacts` (`list_contacts` may persist those fills) so path-table and contact refreshes keep announce aliases.
- **`announce.received`:** coalesced WS notify for LXMF identity announces / path responses (named or nameless). Sidecar applies identity-key + display-name cache updates immediately, but emits **at most one** WS frame per coalesce window (500ms normal / 1000ms when >256 distinct destinations are pending) so announce storms stay O(1) bus pressure on large meshes (~100k). Payload is either a single `{ destination_hash, display_name?, hops, aspect?, identity_hash? }` (legacy / one-row flush) or `{ announces: [{ destination_hash, display_name?, hops, aspect?, identity_hash? }, ...] }` (capped at 1024, named preferred; overflow dropped — slow peer poll recovers). `aspect` is set when announce `name_hash` maps to a known app name (`lxmf.delivery`, `lxmf.propagation`, `nomadnetwork.node`, `rrc.hub`); omitted for path responses / unknown hashes (clients must not invent `"unknown"`). `identity_hash` is the hex identity recovered from the validated announce when present. Each flush publishes pressure counters under `GET /api/v1/diagnostics` → `announce_ws` (ingress/unique/overflow + storm/flush timestamps) for the Diagnostics `reticulum/announce-bus-pressure` warning. Display names update the peer-label cache only — announces do **not** auto-create LXMF contacts. That cache is overlayed onto `GET /api/v1/peers` / topology rows **and** onto nameless/hash-prefix rows from `GET /api/v1/contacts` (`list_contacts` may persist those fills) so path-table and contact refreshes keep announce aliases.
- **`peers_updated`:** also emitted when the live path table **gains** new destination hashes (maintenance tick). Payload may include `{ added: string[], patches: PeerRow[], count }` (added/patches capped at 1024). Renderer applies patches incrementally, including route-field changes. A full peer dump is used on connect, manual Refresh, restart, safety poll, or a `peers_updated` payload that cannot be applied incrementally: `cleared`, `demoted_from_contacts`, or a single-`hash` probe/path event. Hop/timestamp-only churn does not emit.

`lxmf_message` payload fields include `sender_hash`, `text`, `timestamp`, `message_hash`, optional `direction` (`inbound` / `outbound`), optional `delivery_status` (`sending` on optimistic outbound rows), optional `reply_to_hash` / `reply_preview_text` (from LXMF `FIELD_REPLY_TO` / `FIELD_REPLY_QUOTE`), and transport markers `received_via` / `sent_via`. Outbound `sent_via` is **path-table / PacketTap evidence**, not “any local RNode enabled”: atomic values are `rf`, `ble`, `tcp`, or `network`; multi-egress observes join with `+` (e.g. `rf+tcp`, `ble+network`). Inbound `received_via` uses the path-table interface name **matched to local interface config** (same atoms — so a TCP hub named “RNS Testnet” is `tcp`, not `network`). Never use Meshtastic-style `both` for Reticulum.
Expand Down
197 changes: 181 additions & 16 deletions reticulum-sidecar/src/stack/announce_ws_coalesce.rs
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,10 @@ pub struct AnnounceWsRow {
pub destination_hash: String,
pub display_name: Option<String>,
pub hops: u8,
/// Known destination aspect string when `name_hash` maps (e.g. `lxmf.delivery`).
pub aspect: Option<String>,
/// Identity hash recovered from the announce payload (hex), when present.
pub identity_hash: Option<String>,
}

/// Snapshot published for `GET /api/v1/diagnostics` (`announce_ws`).
Expand All @@ -39,6 +43,10 @@ static LAST_WINDOW_OVERFLOW: AtomicU64 = AtomicU64::new(0);
static LAST_STORM_AT_MS: AtomicU64 = AtomicU64::new(0);
static LAST_FLUSH_AT_MS: AtomicU64 = AtomicU64::new(0);

#[cfg(feature = "rns-stack")]
static KNOWN_ANNOUNCE_ASPECTS: std::sync::OnceLock<Vec<([u8; 10], &'static str)>> =
std::sync::OnceLock::new();

fn now_unix_ms() -> u64 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
Expand Down Expand Up @@ -140,29 +148,72 @@ impl AnnounceWsCoalescer {
}
}

/// Map announce `name_hash` (`SHA-256(app_name)[:10]`) to a known aspect string.
///
/// Unknown / zero hashes return `None` — callers must not invent `"unknown"`.
#[cfg(feature = "rns-stack")]
pub fn resolve_announce_aspect(name_hash: &[u8; 10]) -> Option<&'static str> {
if name_hash.iter().all(|&b| b == 0) {
return None;
}
let table = KNOWN_ANNOUNCE_ASPECTS.get_or_init(|| {
[
"lxmf.delivery",
"lxmf.propagation",
"nomadnetwork.node",
"rrc.hub",
]
.into_iter()
.map(|aspect| (rns_identity::name_hash::name_hash(aspect), aspect))
.collect()
});
table
.iter()
.find(|(nh, _)| nh == name_hash)
.map(|(_, aspect)| *aspect)
}

fn announce_row_payload(r: &AnnounceWsRow) -> serde_json::Value {
let mut map = serde_json::Map::new();
map.insert(
"destination_hash".to_string(),
serde_json::Value::String(r.destination_hash.clone()),
);
map.insert(
"display_name".to_string(),
match &r.display_name {
Some(n) => serde_json::Value::String(n.clone()),
None => serde_json::Value::Null,
},
);
map.insert(
"hops".to_string(),
serde_json::Value::Number(serde_json::Number::from(r.hops)),
);
if let Some(ref aspect) = r.aspect {
map.insert(
"aspect".to_string(),
serde_json::Value::String(aspect.clone()),
);
}
if let Some(ref identity_hash) = r.identity_hash {
map.insert(
"identity_hash".to_string(),
serde_json::Value::String(identity_hash.clone()),
);
}
serde_json::Value::Object(map)
}

/// Build the WS text frame for one flush. Single-row keeps the legacy payload shape.
pub fn build_announce_received_frame(rows: &[AnnounceWsRow]) -> Option<String> {
if rows.is_empty() {
return None;
}
let payload = if rows.len() == 1 {
let r = &rows[0];
serde_json::json!({
"destination_hash": r.destination_hash,
"display_name": r.display_name,
"hops": r.hops,
})
announce_row_payload(&rows[0])
} else {
let announces: Vec<serde_json::Value> = rows
.iter()
.map(|r| {
serde_json::json!({
"destination_hash": r.destination_hash,
"display_name": r.display_name,
"hops": r.hops,
})
})
.collect();
let announces: Vec<serde_json::Value> = rows.iter().map(announce_row_payload).collect();
serde_json::json!({ "announces": announces })
};
Some(
Expand Down Expand Up @@ -192,6 +243,23 @@ mod tests {
destination_hash: hash.to_string(),
display_name: name.map(str::to_string),
hops: 1,
aspect: None,
identity_hash: None,
}
}

fn row_full(
hash: &str,
name: Option<&str>,
aspect: Option<&str>,
identity_hash: Option<&str>,
) -> AnnounceWsRow {
AnnounceWsRow {
destination_hash: hash.to_string(),
display_name: name.map(str::to_string),
hops: 2,
aspect: aspect.map(str::to_string),
identity_hash: identity_hash.map(str::to_string),
}
}

Expand All @@ -217,6 +285,31 @@ mod tests {
assert!(snap.last_flush_at_ms > 0);
}

#[test]
fn last_write_wins_preserves_latest_aspect_and_identity_hash() {
let _guard = pressure_metrics_lock();
let mut c = AnnounceWsCoalescer::new();
c.push(row_full(
"aa",
Some("Old"),
Some("lxmf.propagation"),
Some("id_old"),
));
c.push(row_full(
"aa",
Some("New"),
Some("lxmf.delivery"),
Some("id_new"),
));
let flushed = c.take_flush_rows();
let aa = flushed
.iter()
.find(|r| r.destination_hash == "aa")
.expect("aa");
assert_eq!(aa.aspect.as_deref(), Some("lxmf.delivery"));
assert_eq!(aa.identity_hash.as_deref(), Some("id_new"));
}

#[test]
fn flush_prefers_named_when_over_cap_and_records_overflow() {
let _guard = pressure_metrics_lock();
Expand Down Expand Up @@ -272,6 +365,24 @@ mod tests {
assert_eq!(v["type"], "announce.received");
assert_eq!(v["payload"]["destination_hash"], "aa");
assert!(v["payload"].get("announces").is_none());
assert!(v["payload"].get("aspect").is_none());
assert!(v["payload"].get("identity_hash").is_none());
}

#[test]
fn build_frame_single_includes_aspect_and_identity_hash_when_present() {
let id = "aabbccddeeff00112233445566778899";
let frame = build_announce_received_frame(&[row_full(
"aa",
Some("Alice"),
Some("lxmf.delivery"),
Some(id),
)])
.unwrap();
let v: serde_json::Value = serde_json::from_str(&frame).unwrap();
assert_eq!(v["payload"]["aspect"], "lxmf.delivery");
assert_eq!(v["payload"]["identity_hash"], id);
assert_eq!(v["payload"]["hops"], 2);
}

#[test]
Expand All @@ -282,6 +393,37 @@ mod tests {
assert_eq!(v["payload"]["announces"].as_array().unwrap().len(), 2);
}

#[test]
fn build_frame_batch_carries_per_row_aspect_and_identity_hash() {
let frame = build_announce_received_frame(&[
row_full("aa", Some("A"), Some("lxmf.delivery"), Some("id_a")),
row_full("bb", None, Some("nomadnetwork.node"), None),
row("cc", Some("C")),
])
.unwrap();
let v: serde_json::Value = serde_json::from_str(&frame).unwrap();
let announces = v["payload"]["announces"].as_array().unwrap();
assert_eq!(announces.len(), 3);
let aa = announces
.iter()
.find(|a| a["destination_hash"] == "aa")
.expect("aa");
assert_eq!(aa["aspect"], "lxmf.delivery");
assert_eq!(aa["identity_hash"], "id_a");
let bb = announces
.iter()
.find(|a| a["destination_hash"] == "bb")
.expect("bb");
assert_eq!(bb["aspect"], "nomadnetwork.node");
assert!(bb.get("identity_hash").is_none());
let cc = announces
.iter()
.find(|a| a["destination_hash"] == "cc")
.expect("cc");
assert!(cc.get("aspect").is_none());
assert!(cc.get("identity_hash").is_none());
}

#[test]
fn many_distinct_dests_still_one_flush_batch() {
let _guard = pressure_metrics_lock();
Expand All @@ -295,4 +437,27 @@ mod tests {
assert!(c.is_empty());
assert!(build_announce_received_frame(&flushed).is_some());
}

#[cfg(feature = "rns-stack")]
#[test]
fn resolve_announce_aspect_maps_known_aspects() {
assert_eq!(
resolve_announce_aspect(&rns_identity::name_hash::name_hash("lxmf.delivery")),
Some("lxmf.delivery")
);
assert_eq!(
resolve_announce_aspect(&rns_identity::name_hash::name_hash("lxmf.propagation")),
Some("lxmf.propagation")
);
assert_eq!(
resolve_announce_aspect(&rns_identity::name_hash::name_hash("nomadnetwork.node")),
Some("nomadnetwork.node")
);
assert_eq!(
resolve_announce_aspect(&rns_identity::name_hash::name_hash("rrc.hub")),
Some("rrc.hub")
);
assert_eq!(resolve_announce_aspect(&[0u8; 10]), None);
assert_eq!(resolve_announce_aspect(&[1u8; 10]), None);
}
}
4 changes: 3 additions & 1 deletion reticulum-sidecar/src/stack/live.rs
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ use tokio::sync::{RwLock, broadcast};

use super::StackHandle;
use super::announce_ws_coalesce::{
AnnounceWsCoalescer, AnnounceWsRow, build_announce_received_frame,
AnnounceWsCoalescer, AnnounceWsRow, build_announce_received_frame, resolve_announce_aspect,
};
use super::config;
use super::local_rnode_primary;
Expand Down Expand Up @@ -2459,6 +2459,8 @@ impl LiveBridge {
destination_hash: dest_hex,
display_name,
hops: evt.hops,
aspect: resolve_announce_aspect(&evt.name_hash).map(str::to_string),
identity_hash: evt.identity_hash.map(hex::encode),
});
}
() = async {
Expand Down
47 changes: 47 additions & 0 deletions src/main/ipc/reticulum-db-handlers.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -322,6 +322,53 @@ describe('reticulum destination / activity prune IPC', () => {
expect(count).toBe(250);
});

it('named identity-activity upsert clears sibling unknown aspect rows', () => {
const dest = 'aa'.repeat(16);
db!
.prepareOnce(
`INSERT INTO reticulum_identity_activity (destination_hash, aspect, identity_hash, last_seen, hops)
VALUES (?, ?, ?, ?, ?)`,
)
.run(dest, 'unknown', null, Date.now() - 1000, 2);
const upsert = handlers.get('db:upsertReticulumIdentityActivity');
upsert?.(event, {
destination_hash: dest,
aspect: 'lxmf.delivery',
identity_hash: 'bb'.repeat(16),
last_seen: Date.now(),
hops: 1,
});
const rows = db!
.prepareOnce(
'SELECT aspect, identity_hash FROM reticulum_identity_activity WHERE destination_hash = ? ORDER BY aspect',
)
.all(dest) as { aspect: string; identity_hash: string | null }[];
expect(rows).toEqual([{ aspect: 'lxmf.delivery', identity_hash: 'bb'.repeat(16) }]);
});

it('named identity-activity batch upsert clears sibling unknown aspect rows', () => {
const dest = 'cc'.repeat(16);
db!
.prepareOnce(
`INSERT INTO reticulum_identity_activity (destination_hash, aspect, identity_hash, last_seen, hops)
VALUES (?, ?, ?, ?, ?)`,
)
.run(dest, 'unknown', null, Date.now() - 1000, 3);
const result = handlers.get('db:upsertReticulumIdentityActivityBatch')?.(event, [
{
destination_hash: dest,
aspect: 'nomadnetwork.node',
last_seen: Date.now(),
hops: 1,
},
]) as { changes: number };
expect(result.changes).toBe(1);
const rows = db!
.prepareOnce('SELECT aspect FROM reticulum_identity_activity WHERE destination_hash = ?')
.all(dest) as { aspect: string }[];
expect(rows.map((r) => r.aspect)).toEqual(['nomadnetwork.node']);
});

it('upsertReticulumDestination normalizes hash casing into one row', () => {
const upsert = handlers.get('db:upsertReticulumDestination');
const mixed = 'AABBCCDDEEFF00112233445566778899';
Expand Down
17 changes: 17 additions & 0 deletions src/main/ipc/reticulum-db-handlers.ts
Original file line number Diff line number Diff line change
Expand Up @@ -721,6 +721,18 @@ export function registerReticulumDbIpcHandlers({ ipcMain }: ReticulumDbIpcDeps):
last_seen = excluded.last_seen,
hops = COALESCE(excluded.hops, reticulum_identity_activity.hops)`;

const IDENTITY_ACTIVITY_DELETE_UNKNOWN_SQL = `DELETE FROM reticulum_identity_activity
WHERE destination_hash = ? AND aspect = 'unknown'`;

function clearUnknownIdentityActivity(
db: NonNullable<ReturnType<typeof getDbForIpc>>,
destinationHash: string,
aspect: string,
): void {
if (aspect === 'unknown') return;
db.prepareOnce(IDENTITY_ACTIVITY_DELETE_UNKNOWN_SQL).run(destinationHash);
}

function parseIdentityActivityRow(row: unknown): {
destinationHash: string;
aspect: string;
Expand Down Expand Up @@ -762,6 +774,7 @@ export function registerReticulumDbIpcHandlers({ ipcMain }: ReticulumDbIpcDeps):
parsed.lastSeen,
parsed.hops,
);
clearUnknownIdentityActivity(db, parsed.destinationHash, parsed.aspect);
return { changes: 1 };
} catch (err) {
finishDbIpcHandler('db:upsertReticulumIdentityActivity', err);
Expand All @@ -781,9 +794,13 @@ export function registerReticulumDbIpcHandlers({ ipcMain }: ReticulumDbIpcDeps):
}
if (parsed.length === 0) return { changes: 0 };
const stmt = db.prepareOnce(IDENTITY_ACTIVITY_UPSERT_SQL);
const clearUnknown = db.prepareOnce(IDENTITY_ACTIVITY_DELETE_UNKNOWN_SQL);
const run = db.transaction(() => {
for (const p of parsed) {
stmt.run(p.destinationHash, p.aspect, p.identityHash, p.lastSeen, p.hops);
if (p.aspect !== 'unknown') {
clearUnknown.run(p.destinationHash);
}
}
});
run();
Expand Down
3 changes: 2 additions & 1 deletion src/renderer/components/InstantTooltipBubble.tsx
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import type { InstantTooltipPosition } from '@/renderer/lib/instantTooltipPosition';
import { Z_INSTANT_TOOLTIP } from '@/renderer/lib/modalZIndex';

export function InstantTooltipBubble({ text, pos }: { text: string; pos: InstantTooltipPosition }) {
return (
Expand All @@ -9,7 +10,7 @@ export function InstantTooltipBubble({ text, pos }: { text: string; pos: Instant
top: pos.top,
left: pos.left,
transform: pos.below ? 'translate(-50%, 0)' : 'translate(-50%, -100%)',
zIndex: 9999,
zIndex: Z_INSTANT_TOOLTIP,
}}
className="pointer-events-none w-64 rounded border border-gray-600 bg-gray-800 px-2.5 py-1.5 text-xs whitespace-pre-wrap text-gray-200 shadow-lg"
>
Expand Down
Loading
Loading