eric/geth
Watch
1
0
Fork
You've already forked geth
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Use native iroh-blobs for CAS fetches

Browse source
This commit is contained in:
Eric Wendland 2026-05-22 16:10:40 +02:00
commit 72f28224ce
7 changed files with 199 additions and 39 deletions
Download patch file Download diff file Expand all files Collapse all files

1
crates/geth-node/Cargo.toml
View file

@ -34,6 +34,7 @@ geth-store = { path = "../geth-store" }
geth-types = { path = "../geth-types" }
hex.workspace = true
iroh.workspace = true
iroh-blobs.workspace = true
swarm-discovery.workspace = true
[dev-dependencies]

167
crates/geth-node/src/lib.rs
View file

@ -51,6 +51,7 @@ use geth_types::{
AuthOpId, BlobHash, Capability, DeviceId, KeyId, NodeId, PrincipalId, ResourceId, ResourceKind,
ResourceName, SshCertId, SshCertRequestId, UnixMillis, UserId,
};
use iroh::protocol::ProtocolHandler;
use std::collections::{BTreeMap, BTreeSet, VecDeque};
use std::path::{Component, Path, PathBuf};
use std::sync::{Arc, Mutex};
@ -157,6 +158,7 @@ pub struct LocalNode {
pub node_id: String,
pub iroh_status: EndpointStatus,
iroh_endpoint: Arc<Mutex<Option<GethIrohEndpoint>>>,
iroh_blob_store: Arc<Mutex<Option<iroh_blobs::store::fs::FsStore>>>,
runtime: Arc<NodeRuntime>,
}
@ -265,6 +267,7 @@ pub fn init_node(paths: &GethPaths) -> Result<LocalNode, NodeError> {
node_id,
iroh_status: EndpointStatus::scaffolded(),
iroh_endpoint: Arc::new(Mutex::new(None)),
iroh_blob_store: Arc::new(Mutex::new(None)),
runtime: Arc::new(NodeRuntime {
pubsub: Mutex::new(PubsubRuntime::default()),
pipes: Mutex::new(PipeRuntime::default()),
@ -748,7 +751,19 @@ pub async fn handle_request_async(
name,
bearer_secret,
} => document_sync_from_peer(node, &peer_node, &name, bearer_secret).await,
other => handle_request(node, other),
other => {
let response = handle_request(node, other)?;
match &response {
ControlResponse::CasAdded { hash, .. } => {
mirror_local_blob_to_iroh_blobs(node, hash).await?;
}
ControlResponse::CasPrivateAdded { encrypted_hash, .. } => {
mirror_local_blob_to_iroh_blobs(node, encrypted_hash).await?;
}
_ => {}
}
Ok(response)
}
}
}
@ -1275,7 +1290,7 @@ async fn cas_fetch_from_peer(
let conn = endpoint
.endpoint()
.connect(node_addr, geth_iroh::ALPN_CONTROL)
.connect(node_addr.clone(), geth_iroh::ALPN_CONTROL)
.await
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let (mut send, mut recv) = conn
@ -1300,7 +1315,7 @@ async fn cas_fetch_from_peer(
endpoint_id,
hash: response_hash,
size_bytes,
content_base64,
content_base64: _,
allowed,
reason,
nonce: response_nonce,
@ -1319,27 +1334,19 @@ async fn cas_fetch_from_peer(
note,
});
}
let content = content_base64
.ok_or_else(|| NodeError::IrohPeer("peer omitted CAS content".to_owned()))
.and_then(|content| {
base64::engine::general_purpose::STANDARD
.decode(content)
.map_err(|error| NodeError::IrohPeer(error.to_string()))
})?;
if content.len() as u64 != size_bytes {
return Err(NodeError::IrohPeer(format!(
"peer announced {size_bytes} CAS bytes but returned {} bytes",
content.len()
)));
}
let cas = LocalCas::new(node.paths.cas_dir());
let info = cas.add_bytes(&content)?;
let info = fetch_blob_via_iroh_blobs(node, node_addr, response_hash.clone()).await?;
if info.hash != response_hash {
return Err(NodeError::IrohPeer(format!(
"peer returned content hash {} for requested {}",
info.hash, response_hash
)));
}
if info.size_bytes != size_bytes {
return Err(NodeError::IrohPeer(format!(
"peer authorized {size_bytes} CAS bytes but iroh-blobs fetched {} bytes",
info.size_bytes
)));
}
store.record_cas_object(
info.hash.as_str(),
info.size_bytes,
@ -1384,6 +1391,105 @@ async fn cas_fetch_from_peer(
}
}
async fn fetch_blob_via_iroh_blobs(
node: &LocalNode,
node_addr: iroh::EndpointAddr,
hash: BlobHash,
) -> Result<geth_cas::BlobInfo, NodeError> {
let endpoint = node
.iroh_endpoint
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)?
.clone()
.ok_or(NodeError::IrohEndpointUnavailable)?;
let blob_store = iroh_blob_store(node)?.ok_or_else(|| {
NodeError::IrohPeer(
"iroh-blobs store is unavailable; restart the daemon with native Iroh enabled"
.to_owned(),
)
})?;
let iroh_hash = iroh_blob_hash(&hash)?;
let conn = endpoint
.endpoint()
.connect(node_addr, iroh_blobs::ALPN)
.await
.map_err(|error| NodeError::IrohPeer(format!("iroh-blobs connect failed: {error}")))?;
blob_store
.remote()
.fetch(conn, iroh_blobs::HashAndFormat::raw(iroh_hash))
.await
.map_err(|error| NodeError::IrohPeer(format!("iroh-blobs fetch failed: {error}")))?;
let bytes = blob_store
.get_bytes(iroh_hash)
.await
.map_err(|error| NodeError::IrohPeer(format!("iroh-blobs export failed: {error}")))?;
let cas = LocalCas::new(node.paths.cas_dir());
let info = cas.add_bytes(bytes.as_ref())?;
mirror_local_blob_to_iroh_blobs(node, &info.hash).await?;
Ok(info)
}
fn iroh_blob_store(node: &LocalNode) -> Result<Option<iroh_blobs::store::fs::FsStore>, NodeError> {
node.iroh_blob_store
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)
.map(|store| store.clone())
}
fn iroh_blob_hash(hash: &BlobHash) -> Result<iroh_blobs::Hash, NodeError> {
let bytes = hex::decode(hash.as_str()).map_err(|error| {
NodeError::IrohPeer(format!("invalid CAS hash for iroh-blobs: {error}"))
})?;
let bytes: [u8; 32] = bytes
.try_into()
.map_err(|_| NodeError::IrohPeer("invalid CAS hash length for iroh-blobs".to_owned()))?;
Ok(iroh_blobs::Hash::from_bytes(bytes))
}
async fn mirror_local_blob_to_iroh_blobs(
node: &LocalNode,
hash: &BlobHash,
) -> Result<(), NodeError> {
let Some(blob_store) = iroh_blob_store(node)? else {
return Ok(());
};
let content = LocalCas::new(node.paths.cas_dir()).read_bytes(hash)?;
let tag = blob_store
.add_slice(&content)
.await
.map_err(|error| NodeError::IrohPeer(format!("iroh-blobs import failed: {error}")))?;
if tag.hash.to_string() != hash.as_str() {
return Err(NodeError::IrohPeer(format!(
"iroh-blobs imported hash {} but local CAS hash is {}",
tag.hash, hash
)));
}
Ok(())
}
async fn mirror_local_cas_to_iroh_blobs(
paths: &GethPaths,
blob_store: &iroh_blobs::store::fs::FsStore,
) -> Result<usize, NodeError> {
let cas = LocalCas::new(paths.cas_dir());
let mut count = 0;
for blob in cas.list()? {
let content = cas.read_bytes(&blob.hash)?;
let tag = blob_store
.add_slice(&content)
.await
.map_err(|error| NodeError::IrohPeer(format!("iroh-blobs import failed: {error}")))?;
if tag.hash.to_string() != blob.hash.as_str() {
return Err(NodeError::IrohPeer(format!(
"iroh-blobs imported hash {} but local CAS hash is {}",
tag.hash, blob.hash
)));
}
count += 1;
}
Ok(count)
}
async fn cas_root_sync_from_peer(
node: &LocalNode,
peer_node: &str,
@ -4442,6 +4548,16 @@ async fn handle_iroh_control_connection(
.map_err(|error| NodeError::IrohPeer(error.to_string()))?;
let remote_endpoint_id = conn.remote_id().to_string();
let alpn = display_alpn(conn.alpn());
if alpn == display_alpn(iroh_blobs::ALPN) {
let blob_store = iroh_blob_store(&node)?.ok_or_else(|| {
NodeError::IrohPeer("iroh-blobs ALPN accepted but blob store is unavailable".to_owned())
})?;
let protocol = iroh_blobs::BlobsProtocol::new(&blob_store, None);
return protocol
.accept(conn)
.await
.map_err(|error| NodeError::IrohPeer(format!("iroh-blobs accept failed: {error}")));
}
let (mut send, mut recv) = conn
.accept_bi()
.await
@ -9557,11 +9673,26 @@ async fn start_daemon_iroh_endpoint(
let store = Store::open(&node.paths.metadata_db())?;
store.upsert_node_endpoint(endpoint_id, &node.node_id, &node.agent_id, "iroh")?;
}
let blob_store =
iroh_blobs::store::fs::FsStore::load(node.paths.cas_dir().join("iroh-blobs"))
.await
.map_err(|error| {
NodeError::IrohPeer(format!("failed to open iroh-blobs store: {error}"))
})?;
let mirrored = mirror_local_cas_to_iroh_blobs(&node.paths, &blob_store).await?;
tracing::info!(
mirrored_blobs = mirrored,
"native iroh-blobs store is ready"
);
node.iroh_status = status;
*node
.iroh_endpoint
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)? = Some(endpoint.clone());
*node
.iroh_blob_store
.lock()
.map_err(|_| NodeError::RuntimeLockPoisoned)? = Some(blob_store);
Ok(Some(endpoint))
}
Err(error) => {