From e62f911810d074ef5689a63eb50b7ef5a1a47d20 Mon Sep 17 00:00:00 2001 From: archipelago Date: Tue, 28 Jul 2026 21:00:44 -0400 Subject: [PATCH] =?UTF-8?q?fix(mesh):=20Reticulum=20resource=20transfers?= =?UTF-8?q?=20actually=20deliver=20=E2=80=94=204=20root=20causes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit E2E-verified both directions dev-box<->x250 over real RF (5KB image ~60s): 1. Sender daemon never called link.identify() — receiver's get_remote_identity() was None, so every arrived transfer carried an empty source_hash and the Rust side dropped it (now also warns instead of silently vanishing it). 2. Receiver treated resource.data as bytes, but RNS hands a concluded Resource's data as a file-like BufferedReader — b64encode raised TypeError and the transfer was lost even when attributed. 3. Radio twins of merged contacts carry the peer's Archipelago ed25519 key as pubkey_hex, not an RNS hash — prefix lookup could never match ('Unknown Reticulum prefix', observed live). resolve_dest_hash now falls back to matching the announce-bound arch_pubkey_hex. 4. The daemon RPC socket kept asyncio's default 64KiB line limit; any attachment >~48KB overflowed it and tore down the whole daemon connection ('reticulum-daemon is gone'). Raised to 16MiB. Co-Authored-By: Claude Fable 5 --- core/archipelago/src/mesh/reticulum.rs | 85 ++++++++++++++++---------- reticulum-daemon/reticulum_daemon.py | 50 ++++++++++++--- 2 files changed, 92 insertions(+), 43 deletions(-) diff --git a/core/archipelago/src/mesh/reticulum.rs b/core/archipelago/src/mesh/reticulum.rs index 1ffb6d85..ae4c8ff5 100644 --- a/core/archipelago/src/mesh/reticulum.rs +++ b/core/archipelago/src/mesh/reticulum.rs @@ -525,6 +525,28 @@ impl ReticulumLink { info!(count = self.peers.len(), "Loaded persisted Reticulum peers"); } + /// Resolve a caller-supplied 6-byte routing prefix to a full 16-byte RNS + /// destination hash. Two shapes arrive here depending on how the contact + /// record was born (observed live 2026-07-28, image-to-merged-contact): + /// 1. the peer's RNS dest-hash prefix (contacts created from an RNS + /// announce) — direct `prefix_to_hash` hit; + /// 2. the peer's Archipelago ed25519 pubkey prefix (radio twins bound + /// via the ARCHY announce blob store the arch key as `pubkey_hex`) — + /// no dest-hash match possible, so fall back to scanning peers whose + /// announce-bound `arch_pubkey_hex` starts with the prefix. + fn resolve_dest_hash(&self, prefix: &[u8; 6]) -> Option<[u8; 16]> { + if let Some(hash) = self.prefix_to_hash.get(prefix) { + return Some(*hash); + } + let hex_prefix = hex::encode(prefix); + self.peers.values().find_map(|p| { + p.arch_pubkey_hex + .as_deref() + .filter(|arch| arch.starts_with(&hex_prefix)) + .map(|_| p.dest_hash) + }) + } + /// Best-effort sync write of the current peer table — called after any /// insert that adds/renames a peer. Infrequent (announces/first-contact, /// not per-message) so a blocking write here is a fine trade for keeping @@ -601,16 +623,12 @@ impl ReticulumLink { dest_pubkey_prefix: &[u8; 6], payload: &[u8], ) -> Result<()> { - let dest_hash = self - .prefix_to_hash - .get(dest_pubkey_prefix) - .copied() - .with_context(|| { - format!( - "Unknown Reticulum prefix {} — peer hasn't announced yet", - hex::encode(dest_pubkey_prefix) - ) - })?; + let dest_hash = self.resolve_dest_hash(dest_pubkey_prefix).with_context(|| { + format!( + "Unknown Reticulum prefix {} — peer hasn't announced yet", + hex::encode(dest_pubkey_prefix) + ) + })?; // Typed-envelope payloads (ReadReceipt/Reaction/etc. — anything small // enough for the single-frame path) are raw binary CBOR, not text. // `from_utf8_lossy` would irreversibly mangle them since `content` @@ -645,16 +663,12 @@ impl ReticulumLink { caption: Option<&str>, ) -> Result<()> { use base64::{engine::general_purpose::STANDARD as B64, Engine as _}; - let dest_hash = self - .prefix_to_hash - .get(dest_pubkey_prefix) - .copied() - .with_context(|| { - format!( - "Unknown Reticulum prefix {} — peer hasn't announced yet", - hex::encode(dest_pubkey_prefix) - ) - })?; + let dest_hash = self.resolve_dest_hash(dest_pubkey_prefix).with_context(|| { + format!( + "Unknown Reticulum prefix {} — peer hasn't announced yet", + hex::encode(dest_pubkey_prefix) + ) + })?; self.send_rpc(serde_json::json!({ "cmd": "send", "dest_hash": hex::encode(dest_hash), @@ -677,16 +691,12 @@ impl ReticulumLink { /// `handle_event`, not awaited here. pub async fn send_resource(&mut self, dest_pubkey_prefix: &[u8; 6], data: &[u8]) -> Result<()> { use base64::{engine::general_purpose::STANDARD as B64, Engine as _}; - let dest_hash = self - .prefix_to_hash - .get(dest_pubkey_prefix) - .copied() - .with_context(|| { - format!( - "Unknown Reticulum prefix {} — peer hasn't announced yet", - hex::encode(dest_pubkey_prefix) - ) - })?; + let dest_hash = self.resolve_dest_hash(dest_pubkey_prefix).with_context(|| { + format!( + "Unknown Reticulum prefix {} — peer hasn't announced yet", + hex::encode(dest_pubkey_prefix) + ) + })?; let req_id = self.next_resource_id(); self.send_rpc(serde_json::json!({ "cmd": "send_resource", @@ -1023,10 +1033,19 @@ impl ReticulumLink { .push_back(build_synthetic_frame(&prefix, &content)); } Some("resource_recv") => { - let Some(source_hex) = ev.get("source_hash").and_then(Value::as_str) else { - return; - }; + let source_hex = ev + .get("source_hash") + .and_then(Value::as_str) + .unwrap_or_default(); let Ok(source_hash) = parse_hash16(source_hex) else { + // An empty source_hash means the sender's link wasn't + // identified (pre-identify daemon build on the far end) — + // the blob is undeliverable without attribution, but say + // so instead of vanishing it. + warn!( + source = source_hex, + "Dropping inbound Reticulum resource without valid source identity" + ); return; }; let prefix: [u8; 6] = source_hash[..6].try_into().unwrap(); diff --git a/reticulum-daemon/reticulum_daemon.py b/reticulum-daemon/reticulum_daemon.py index 65e8dc63..02d38d3f 100644 --- a/reticulum-daemon/reticulum_daemon.py +++ b/reticulum-daemon/reticulum_daemon.py @@ -437,13 +437,29 @@ class ReticulumDaemon: identity, RNS.Destination.OUT, RNS.Destination.SINGLE, "archy", "resource", ) - link = RNS.Link(out_dest) + # established_callback identifies us to the peer BEFORE any Resource + # goes out: the receiving daemon attributes an inbound resource via + # link.get_remote_identity() (see _on_resource_received), which is + # None unless the initiator calls identify() — without this every + # transfer arrives with an empty source_hash and the Rust side can't + # route it to a contact (observed live 2026-07-28: 5KB image sent, + # concluded COMPLETE, never surfaced on the receiving node). + link = RNS.Link( + out_dest, + established_callback=lambda lk: self._on_out_link_established( + resource_hash, lk + ), + ) link.set_link_closed_callback( lambda lk: self.links.pop(resource_hash, None) ) self.links[resource_hash] = link return link, resource_hash + def _on_out_link_established(self, resource_hash: bytes, link): + link.identify(self.identity) + self._flush_pending_resource_sends(resource_hash, link) + def _send_resource(self, req: dict): import RNS req_id = req.get("id", "") @@ -463,14 +479,14 @@ class ReticulumDaemon: if link.status == RNS.Link.ACTIVE: self._start_resource(link, data, req_id) else: - # Link is establishing — queue and flush from the established - # callback. set_link_established_callback only fires once per Link - # object (the cache is reused across sends), so re-set it here to - # make sure THIS send's queue entry gets flushed too. + # Link is establishing — queue; the creation-time established + # callback (_on_out_link_established) identifies us then flushes + # the queue. Re-check afterwards to close the race where the + # link went ACTIVE between the status check and the append + # (the callback fires once, on the RNS thread). self.pending_resource_sends[resource_hash].append((data, req_id)) - link.set_link_established_callback( - lambda lk: self._flush_pending_resource_sends(resource_hash, lk) - ) + if link.status == RNS.Link.ACTIVE: + self._flush_pending_resource_sends(resource_hash, link) def _flush_pending_resource_sends(self, resource_hash: bytes, link): import RNS @@ -518,10 +534,17 @@ class ReticulumDaemon: RNS.Destination.hash(identity, "lxmf", "delivery").hex() if identity is not None else "" ) + # RNS hands a concluded Resource's data as a file-like object + # (BufferedReader over the assembled stream), not bytes — + # observed live 2026-07-28: b64encode raised "a bytes-like + # object is required" and every received transfer was lost. + data = resource.data + if hasattr(data, "read"): + data = data.read() self._emit_threadsafe({ "event": "resource_recv", "source_hash": source_hash, - "data_b64": base64.b64encode(resource.data).decode("ascii"), + "data_b64": base64.b64encode(data).decode("ascii"), }) except Exception as e: # never let a callback kill the RNS thread self._emit_threadsafe({"event": "error", "where": "resource_recv", @@ -531,7 +554,14 @@ class ReticulumDaemon: sock_path = self.args.socket if os.path.exists(sock_path): os.unlink(sock_path) - server = await asyncio.start_unix_server(self._handle_client, path=sock_path) + # limit: asyncio's default per-line StreamReader cap is 64KiB, but + # send_resource requests carry the whole payload base64-encoded on one + # JSON line — a ~48KB+ attachment overflows the default and the raised + # LimitOverrunError tears down the RPC connection (the Rust side then + # sees "reticulum-daemon is gone" and restarts the whole session). + server = await asyncio.start_unix_server( + self._handle_client, path=sock_path, limit=16 * 1024 * 1024 + ) os.chmod(sock_path, 0o600) self.announce() async with server: