fix(mesh): Reticulum resource transfers actually deliver — 4 root causes
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 <noreply@anthropic.com>
This commit is contained in:
parent
79c3cc5947
commit
e62f911810
@ -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();
|
||||
|
||||
@ -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:
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user