core: M4 — files/photos over iroh-blobs (multi-modal clipboard)

IrohTransport.send_file now adds the (already-sealed) bytes to a MemStore and
broadcasts a blob announcement over gossip {hash, name, mime, provider}; on
receipt, peers fetch the blob P2P by hash via iroh-blobs and emit MeshEvent::File.
The engine's existing File handler decrypts and delivers — so the clipboard can
carry data/photos/etc, not just text. Replaces the hand-rolled M/C/E frame
protocol entirely.

Verified: `cargo run --features iroh-transport --example iroh_engine` syncs both
a text clipboard (gossip) AND a 200 KB photo (iroh-blobs), both E2E-decrypted
A→B, no SSE/RTC/LAN/relay. Default ship build untouched; feature build clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-22 01:15:41 -05:00
parent b377a6708a
commit 3cb916ee5e
5 changed files with 706 additions and 81 deletions

View File

@@ -430,10 +430,16 @@ multi-modal payloads (text inline, audio/photo/data via `iroh-blobs`). Receiving
--example iroh_engine` → the **real Engine** (unchanged public API)
syncs a clipboard A→B, E2E-decrypted, **no SSE/RTC/LAN/relay**. The
seam paid off — zero engine/client changes.
- [x] **Files via `iroh-blobs`**`send_file` adds the (sealed) bytes
to a `MemStore`, broadcasts a blob announcement over gossip;
peers fetch P2P by hash and emit `MeshEvent::File`. Replaces the
M/C/E frame protocol. Verified: a 200 KB photo syncs A→B
E2E-decrypted alongside text (`iroh_engine` example).
- [ ] Remaining: rendezvous for zero-config bootstrap (today via
`TETHER_IROH_BOOTSTRAP` env stopgap); file transfer via
`iroh-blobs`; make webrtc/mdns/reqwest optional so the iroh/wasm
build is lean; A/B vs RTC+LAN; flip the feature on by default.
`TETHER_IROH_BOOTSTRAP` env stopgap); cross-compile the feature
for iOS + build the xcframework with it; make webrtc/mdns/reqwest
optional so the iroh/wasm build is lean; A/B vs RTC+LAN; flip the
feature on by default.
- [ ] Spike **Automerge** as the Sync tier; re-express the clipboard as a tiny
app over replicated state to validate the App API.
- [ ] Durable device keypairs + authenticated handshake (Noise/`snow`, or

593
core/Cargo.lock generated

File diff suppressed because it is too large Load Diff

View File

@@ -34,10 +34,11 @@ getrandom = "0.4.3"
# the hand-rolled SSE/RTC/LAN stack.
iroh = { version = "1", optional = true }
iroh-gossip = { version = "0.101", optional = true }
iroh-blobs = { version = "0.103", optional = true }
[features]
default = []
iroh-transport = ["dep:iroh", "dep:iroh-gossip"]
iroh-transport = ["dep:iroh", "dep:iroh-gossip", "dep:iroh-blobs"]
[dev-dependencies]
# Iroh spikes (examples/iroh_*.rs) link iroh directly regardless of the feature.

View File

@@ -22,13 +22,18 @@ impl MessageHandler for Noop {
fn on_file(&self, _n: String, _m: String, _d: Vec<u8>) {}
}
struct Collector(mpsc::Sender<String>);
struct Collector {
text: mpsc::Sender<String>,
file: mpsc::Sender<(String, usize)>,
}
impl MessageHandler for Collector {
fn on_message(&self, m: Message) {
let _ = self.0.send(m.text);
let _ = self.text.send(m.text);
}
fn on_status(&self, _c: bool) {}
fn on_file(&self, _n: String, _m: String, _d: Vec<u8>) {}
fn on_file(&self, name: String, _mime: String, data: Vec<u8>) {
let _ = self.file.send((name, data.len()));
}
}
fn main() {
@@ -61,28 +66,52 @@ fn main() {
// Joiner (bootstraps to A via the env var its IrohTransport reads at start).
let (tx, rx) = mpsc::channel();
let (ftx, frx) = mpsc::channel();
let b = Engine::new(
server.into(), room.into(), "iroh-b".into(), "ios".into(), "B".into(), account.into(),
);
b.start(Box::new(Collector(tx)));
b.start(Box::new(Collector { text: tx, file: ftx }));
eprintln!("waiting for gossip topic to form…");
std::thread::sleep(Duration::from_secs(8));
// 1) Text clipboard over gossip.
let payload = "clipboard synced over iroh 🎉";
eprintln!("A sending: {payload:?}");
eprintln!("A sending text: {payload:?}");
a.send(payload.to_string());
match rx.recv_timeout(Duration::from_secs(10)) {
Ok(t) if t == payload => {
println!("engine-over-iroh: B received {t:?} (E2E-decrypted, via gossip, no relay/SSE/RTC/LAN)");
println!("text: B received {t:?} (E2E-decrypted, via gossip)");
}
Ok(t) => {
println!("⚠️ B got {t:?} (expected {payload:?})");
std::process::exit(1);
}
Ok(t) => println!("⚠️ B got {t:?} (expected {payload:?})"),
Err(_) => {
println!("❌ B received nothing — gossip topic may not have formed");
println!("❌ B received no text");
std::process::exit(1);
}
}
// 2) A "photo"/blob over iroh-blobs (announced via gossip, fetched P2P).
let photo: Vec<u8> = (0..200_000u32).map(|i| (i % 251) as u8).collect();
eprintln!("A sending blob: photo.bin ({} bytes)", photo.len());
a.send_file("photo.bin".into(), "image/png".into(), photo.clone());
match frx.recv_timeout(Duration::from_secs(15)) {
Ok((name, len)) if len == photo.len() => {
println!("✅ blob: B received {name:?} ({len} bytes, E2E-decrypted, via iroh-blobs)");
}
Ok((name, len)) => {
println!("⚠️ B got {name:?} with {len} bytes (expected {})", photo.len());
std::process::exit(1);
}
Err(_) => {
println!("❌ B received no blob");
std::process::exit(1);
}
}
println!("🎉 engine-over-iroh: text + photo synced, no SSE/RTC/LAN/relay");
a.stop();
b.stop();
}

View File

@@ -1,7 +1,8 @@
//! IrohTransport — the Mesh, on iroh. The end state of M4: this one transport
//! subsumes the hand-rolled SSE/RTC/LAN stack. iroh handles key-addressing,
//! holepunch, and relay; `iroh-gossip` carries the clipboard `Message`s and
//! presence over the account-derived topic.
//! presence; `iroh-blobs` carries files/photos/data (content-addressed, fetched
//! P2P on demand) — replacing the hand-rolled M/C/E frame protocol.
//!
//! It implements the same `Transport` trait and emits the same `MeshEvent`s as
//! the legacy transports, so the engine event loop and every client are
@@ -9,38 +10,64 @@
//!
//! Bootstrap note (v1): joining the gossip topic needs ≥1 peer id. Until the
//! rendezvous supernode lands (see ARCHITECTURE §M3), bootstrap ids come from
//! the `TETHER_IROH_BOOTSTRAP` env var (comma-separated). That's the one
//! stopgap; everything else is production-shaped.
//! the `TETHER_IROH_BOOTSTRAP` env var (comma-separated).
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex as StdMutex};
use futures_util::TryStreamExt;
use iroh::{endpoint::presets, protocol::Router, Endpoint, EndpointId};
use iroh_blobs::{store::mem::MemStore, BlobsProtocol, Hash};
use iroh_gossip::{
api::{Event, GossipSender},
net::{Gossip, GOSSIP_ALPN},
proto::TopicId,
};
use serde::{Deserialize, Serialize};
use tokio::runtime::Handle;
use crate::{Config, EventTx, MeshEvent, Message, Transport};
/// Blob handles needed by `send_file` / blob downloads, set once online.
#[derive(Clone)]
struct Blobs {
store: MemStore,
endpoint: Endpoint,
my_id: String,
}
pub(crate) struct IrohTransport {
cfg: Config,
events: EventTx,
/// Set once the gossip topic is joined; used by `publish`.
sender: Arc<StdMutex<Option<GossipSender>>>,
/// Blob store + endpoint, set once online; used by `send_file` and downloads.
blobs: Arc<StdMutex<Option<Blobs>>>,
/// This node's iroh EndpointId, surfaced once online (for bootstrap wiring).
id: Arc<StdMutex<Option<String>>>,
}
impl IrohTransport {
pub(crate) fn new(cfg: Config, events: EventTx, id: Arc<StdMutex<Option<String>>>) -> Self {
Self { cfg, events, sender: Arc::new(StdMutex::new(None)), id }
Self {
cfg,
events,
sender: Arc::new(StdMutex::new(None)),
blobs: Arc::new(StdMutex::new(None)),
id,
}
}
}
/// A blob announcement, broadcast over gossip so peers can fetch the bytes P2P.
#[derive(Serialize, Deserialize)]
struct BlobAnnounce {
hash: String, // blake3 hash of the (sealed) bytes
name: String,
mime: String,
provider: String, // the sender's iroh EndpointId to download from
}
/// account/room → 32-byte gossip topic (consistent with room_for_account: the
/// room is already sha256(account)[..4], this is the full-width derivation).
fn topic_for(room: &str) -> TopicId {
@@ -62,6 +89,34 @@ fn bootstrap_ids() -> Vec<EndpointId> {
.collect()
}
impl IrohTransport {
/// Fetch a blob announced over gossip from its provider and surface it.
async fn fetch_blob(blobs: Blobs, events: EventTx, ann: BlobAnnounce) {
let (Ok(hash), Ok(provider)) =
(ann.hash.parse::<Hash>(), ann.provider.parse::<EndpointId>())
else {
return;
};
let downloader = blobs.store.downloader(&blobs.endpoint);
if let Err(e) = downloader.download(hash, Some(provider)).await {
eprintln!("tethercore[iroh]: blob download failed: {e}");
return;
}
match blobs.store.get_bytes(hash).await {
Ok(bytes) => {
eprintln!("tethercore[iroh]: fetched blob {:?} ({} bytes)", ann.name, bytes.len());
// Bytes are still sealed; the engine's File handler decrypts.
let _ = events.send(MeshEvent::File {
name: ann.name,
mime: ann.mime,
data: bytes.to_vec(),
});
}
Err(e) => eprintln!("tethercore[iroh]: blob read failed: {e}"),
}
}
}
impl Transport for IrohTransport {
fn name(&self) -> &'static str {
"iroh"
@@ -87,9 +142,17 @@ impl Transport for IrohTransport {
eprintln!("tethercore[iroh]: online id={my_id}");
let gossip = Gossip::builder().spawn(endpoint.clone());
let store = MemStore::new();
let blobs_proto = BlobsProtocol::new(&store, None);
let router = Router::builder(endpoint.clone())
.accept(GOSSIP_ALPN, gossip.clone())
.accept(iroh_blobs::ALPN, blobs_proto)
.spawn();
*this.blobs.lock().unwrap() = Some(Blobs {
store: store.clone(),
endpoint: endpoint.clone(),
my_id: my_id.clone(),
});
let topic = topic_for(&this.cfg.room);
let sub = match gossip.subscribe_and_join(topic, bootstrap_ids()).await {
@@ -107,10 +170,22 @@ impl Transport for IrohTransport {
while running.load(Ordering::SeqCst) {
match receiver.try_next().await {
Ok(Some(Event::Received(msg))) => {
if let Ok(m) = serde_json::from_slice::<Message>(&msg.content) {
if m.from != this.cfg.from {
let _ = this.events.send(MeshEvent::Message(m));
let Ok(m) = serde_json::from_slice::<Message>(&msg.content) else {
continue;
};
if m.from == this.cfg.from {
continue; // our own broadcast
}
if m.kind == "blob" {
// File announcement → fetch the bytes P2P, then emit File.
if let Ok(ann) = serde_json::from_str::<BlobAnnounce>(&m.text) {
if let Some(blobs) = this.blobs.lock().unwrap().clone() {
let events = this.events.clone();
tokio::spawn(Self::fetch_blob(blobs, events, ann));
}
}
} else {
let _ = this.events.send(MeshEvent::Message(m));
}
}
Ok(Some(Event::NeighborUp(id))) => {
@@ -119,8 +194,8 @@ impl Transport for IrohTransport {
Ok(Some(Event::NeighborDown(id))) => {
let _ = this.events.send(MeshEvent::DirectDown(id.to_string()));
}
Ok(Some(_)) => {} // Lagged, etc.
Ok(None) => break, // stream ended
Ok(Some(_)) => {} // Lagged, etc.
Ok(None) => break, // stream ended
Err(e) => {
eprintln!("tethercore[iroh]: recv error: {e}");
break;
@@ -147,4 +222,43 @@ impl Transport for IrohTransport {
});
}
}
/// Add the (already-sealed) bytes to the local blob store and broadcast a
/// blob announcement over gossip; peers fetch the bytes P2P by hash.
fn send_file(&self, rt: &Handle, _id: String, name: String, mime: String, data: Vec<u8>) {
let blobs = self.blobs.lock().unwrap().clone();
let sender = self.sender.lock().unwrap().clone();
let cfg = self.cfg.clone();
let (Some(blobs), Some(sender)) = (blobs, sender) else {
return;
};
rt.spawn(async move {
let tag = match blobs.store.add_slice(&data).await {
Ok(t) => t,
Err(e) => {
eprintln!("tethercore[iroh]: blob add failed: {e}");
return;
}
};
let ann = BlobAnnounce {
hash: tag.hash.to_string(),
name,
mime,
provider: blobs.my_id.clone(),
};
let msg = Message {
kind: "blob".into(),
text: serde_json::to_string(&ann).unwrap_or_default(),
from: cfg.from.clone(),
to: String::new(),
role: String::new(),
source: cfg.source.clone(),
room: cfg.room.clone(),
ts: 0,
};
if let Ok(body) = serde_json::to_vec(&msg) {
let _ = sender.broadcast(body.into()).await;
}
});
}
}