greg/synchronicity
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 10 Wiki Activity

p2p: trim protocol-handler boilerplate

Browse Source 
EnsembleProtocol gains an associated Error type, so each handler's serve
is its former run body and Gated converts with AcceptError::from_err —
instead of six serve shims that flattened the error to a string and lost
its source chain. The Debug supertrait is dropped (Gated prints the
handler's type name), taking eight hand-written Debug impls and six new()
constructors with it; Retiring derives Debug and uses Default.

A small Gate struct replaces the local generic gate() fn, which shrinks
P2PNode::start's 60-line router block to one line per ALPN. ClientIntent,
generic but used by artifact-sync alone, becomes a plain enum there. The
admission lint's accepted spelling follows the rename.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Greg Shuflin
2026-09-28 05:40:07 -07:00
co-authored by Claude Opus 5.5
parent 5463d8a6dc
commit ae10b2b6d9
9 changed files with 113 additions and 312 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rust/lib/src/p2p/admission.rs
+27 -27
View File
@@ -230,21 +230,10 @@ pub enum RevokedPeers {
/// are the wipe and killing the process, and the second leaves nothing behind.
///
/// See `docs/NODE_REVOCATION.md` §2.
#[derive(Clone, Default)]
#[derive(Debug, Clone, Default)]
pub struct Retiring(Arc<AtomicBool>);
impl std::fmt::Debug for Retiring {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_tuple("Retiring").field(&self.is_retiring()).finish()
}
}
impl Retiring {
/// A node that is not retiring — the state every node starts in.
pub fn new() -> Self {
Self::default()
}
/// Enter the retiring state. There is no leaving it.
pub fn begin(&self) {
self.0.store(true, Ordering::SeqCst);
@@ -410,7 +399,9 @@ fn check_self_serving(
/// succeeded for `conn`; use it in preference to `conn.remote_id()` when
/// recording who a write came from, since the two are guaranteed equal and
/// the former documents that it was checked.
pub trait EnsembleProtocol: std::fmt::Debug + Send + Sync + 'static {
pub trait EnsembleProtocol: Send + Sync + 'static {
type Error: std::error::Error + Send + Sync + 'static;
/// Whether this protocol serves a peer whose cert is revoked, and is
/// served while *this* node's own cert is revoked.
///
@@ -424,7 +415,7 @@ pub trait EnsembleProtocol: std::fmt::Debug + Send + Sync + 'static {
&self,
conn: Connection,
peer: CertifiedPeer,
) -> impl std::future::Future<Output = std::result::Result<(), AcceptError>> + Send;
) -> impl std::future::Future<Output = std::result::Result<(), Self::Error>> + Send;
}
/// Wraps an [`EnsembleProtocol`] into an iroh `ProtocolHandler` that runs the
@@ -436,15 +427,22 @@ pub struct Gated<H> {
retiring: Retiring,
}
impl<H: std::fmt::Debug> std::fmt::Debug for Gated<H> {
impl<H> std::fmt::Debug for Gated<H> {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Gated").field("inner", &self.inner).finish()
f.debug_tuple("Gated").field(&std::any::type_name::<H>()).finish()
}
}
impl<H: EnsembleProtocol> Gated<H> {
pub fn new(inner: H, store: Arc<Store>, retiring: Retiring) -> Self {
Gated { inner, store, retiring }
/// What every gated handler on one node shares. [`gate`](Self::gate) is the
/// only way to build a [`Gated`].
pub struct Gate {
pub store: Arc<Store>,
pub retiring: Retiring,
}
impl Gate {
pub fn gate<H: EnsembleProtocol>(&self, inner: H) -> Gated<H> {
Gated { inner, store: Arc::clone(&self.store), retiring: self.retiring.clone() }
}
}
@@ -460,7 +458,7 @@ impl<H: EnsembleProtocol> ProtocolHandler for Gated<H> {
return Ok(());
}
};
self.inner.serve(conn, peer).await
self.inner.serve(conn, peer).await.map_err(AcceptError::from_err)
}
}
@@ -881,11 +879,12 @@ mod tests {
deny(&store, &RevocationRecord::self_signed(&us.cert, &us.keypair));
let membership = membership_for(&us.cert);
let err = check_self_serving(&store, &membership, &Retiring::new(), RevokedPeers::Refuse)
.expect_err("a revoked node must not serve or dial");
let err =
check_self_serving(&store, &membership, &Retiring::default(), RevokedPeers::Refuse)
.expect_err("a revoked node must not serve or dial");
assert!(matches!(err, AdmissionError::SelfRevoked));
check_self_serving(&store, &membership, &Retiring::new(), RevokedPeers::Admit)
check_self_serving(&store, &membership, &Retiring::default(), RevokedPeers::Admit)
.expect("except to deliver its own revocation");
}
@@ -946,14 +945,15 @@ mod tests {
let membership = membership_for(&us.cert);
bump_to(&store, &ens, 3);
let err = check_self_serving(&store, &membership, &Retiring::new(), RevokedPeers::Refuse)
.expect_err("a stale node must not serve or dial");
let err =
check_self_serving(&store, &membership, &Retiring::default(), RevokedPeers::Refuse)
.expect_err("a stale node must not serve or dial");
assert!(
matches!(err, AdmissionError::SelfStaleEpoch { cert_epoch: 0, current_epoch: 3 }),
"{err}"
);
check_self_serving(&store, &membership, &Retiring::new(), RevokedPeers::Admit)
check_self_serving(&store, &membership, &Retiring::default(), RevokedPeers::Admit)
.expect("except to be handed the bump that explains it");
}
@@ -966,7 +966,7 @@ mod tests {
let ens = ensemble_kp();
let us = Peer::new(&ens);
let membership = membership_for(&us.cert);
let retiring = Retiring::new();
let retiring = Retiring::default();
check_self_serving(&store, &membership, &retiring, RevokedPeers::Refuse)
.expect("a node that has not begun retiring serves normally");
rust/lib/src/p2p/artifact_sync.rs
+13 -32
View File
@@ -32,16 +32,13 @@
use std::sync::Arc;
use iroh::{
EndpointAddr, EndpointId, KeyParsingError, endpoint::Connection, protocol::AcceptError,
};
use iroh::{EndpointAddr, EndpointId, KeyParsingError, endpoint::Connection};
use log::info;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::documents::{Artifact, ArtifactKind, ArtifactRef, ArtifactValue, DocumentID};
use crate::node::NodePubkey;
use crate::p2p::ClientIntent;
use crate::p2p::admission::{CertifiedPeer, EnsembleEndpoint, EnsembleProtocol};
use crate::p2p::wire::{self, WireError, read_msg, write_msg};
use crate::p2p::wire_time::{SkewedStamps, wire_timestamp};
@@ -49,9 +46,13 @@ use crate::stores::Store;
use crate::stores::artifact::ArtifactStoreError;
use crate::stores::node_content_status::{ContentKey, NodeContentStatusError};
/// This protocol's fetch request and push batch are different shapes, so the
/// two arms of [`ClientIntent`] here carry different payload types.
type ArtifactClientIntent = ClientIntent<ArtifactSyncRequest, ArtifactPushBatch>;
/// The first message on a session: whether the connecting node wants to fetch
/// values from the responder, or push values to it.
#[derive(Debug, Serialize, Deserialize)]
enum ArtifactClientIntent {
Fetch(ArtifactSyncRequest),
Push(ArtifactPushBatch),
}
/// Bumped to `/2` when note bodies became CRDT documents
/// (`docs/adr/0030-note-body-crdt-loro.md`). That change reshaped `NoteBody`,
@@ -170,35 +171,13 @@ pub struct ArtifactSyncResult {
/// Register with `Router::accept(ARTIFACT_SYNC_ALPN, handler)`.
#[derive(Clone)]
pub struct ArtifactSyncHandler {
store: Arc<Store>,
}
impl ArtifactSyncHandler {
pub fn new(store: Arc<Store>) -> Self {
Self { store }
}
}
impl std::fmt::Debug for ArtifactSyncHandler {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("ArtifactSyncHandler").field("store_path", &self.store.path).finish()
}
pub(crate) store: Arc<Store>,
}
impl EnsembleProtocol for ArtifactSyncHandler {
async fn serve(
&self,
conn: Connection,
peer: CertifiedPeer,
) -> std::result::Result<(), AcceptError> {
self.run(conn, peer)
.await
.map_err(|e| AcceptError::from_err(std::io::Error::other(e.to_string())))
}
}
type Error = ArtifactSyncError;
impl ArtifactSyncHandler {
async fn run(&self, conn: Connection, peer: CertifiedPeer) -> Result<()> {
async fn serve(&self, conn: Connection, peer: CertifiedPeer) -> Result<()> {
let remote = conn.remote_id();
let (mut send, mut recv) = wire::accept(&conn).await?;
@@ -214,7 +193,9 @@ impl ArtifactSyncHandler {
conn.closed().await;
Ok(())
}
}
impl ArtifactSyncHandler {
/// Answer a `Fetch`: send back the value of each requested artifact we hold.
async fn serve_fetch(
&self,
rust/lib/src/p2p/blob_stream.rs
+6 -28
View File
@@ -51,9 +51,7 @@
use std::sync::Arc;
use iroh::{
EndpointAddr, EndpointId, KeyParsingError, endpoint::Connection, protocol::AcceptError,
};
use iroh::{EndpointAddr, EndpointId, KeyParsingError, endpoint::Connection};
use log::info;
use serde::{Deserialize, Serialize};
@@ -173,35 +171,13 @@ fn trim_chunk_bounds(
/// Register with `Router::accept(BLOB_STREAM_ALPN, handler)`.
#[derive(Clone)]
pub struct BlobStreamHandler {
store: Arc<Store>,
}
impl BlobStreamHandler {
pub fn new(store: Arc<Store>) -> Self {
Self { store }
}
}
impl std::fmt::Debug for BlobStreamHandler {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("BlobStreamHandler").field("store_path", &self.store.path).finish()
}
pub(crate) store: Arc<Store>,
}
impl EnsembleProtocol for BlobStreamHandler {
async fn serve(
&self,
conn: Connection,
_peer: CertifiedPeer,
) -> std::result::Result<(), AcceptError> {
self.run(conn)
.await
.map_err(|e| AcceptError::from_err(std::io::Error::other(e.to_string())))
}
}
type Error = BlobStreamError;
impl BlobStreamHandler {
async fn run(&self, conn: Connection) -> Result<()> {
async fn serve(&self, conn: Connection, _peer: CertifiedPeer) -> Result<()> {
let remote = conn.remote_id();
let (mut send, mut recv) = wire::accept(&conn).await?;
@@ -212,7 +188,9 @@ impl BlobStreamHandler {
conn.closed().await;
Ok(())
}
}
impl BlobStreamHandler {
async fn serve_request(
&self,
remote: EndpointId,
rust/lib/src/p2p/blob_sync.rs
+7 -32
View File
@@ -40,9 +40,7 @@
use std::sync::Arc;
use iroh::{
EndpointAddr, EndpointId, KeyParsingError, endpoint::Connection, protocol::AcceptError,
};
use iroh::{EndpointAddr, EndpointId, KeyParsingError, endpoint::Connection};
use log::info;
use serde::{Deserialize, Serialize};
@@ -57,10 +55,7 @@ use crate::stores::blob_store::BlobStoreError;
use crate::stores::document::DocumentStoreError;
use crate::stores::node_content_status::ContentKey;
/// This protocol's client-initiated requests. Unlike [`artifact_sync`](super::artifact_sync)
/// and doc_sync's two-armed [`crate::p2p::ClientIntent`], this module needs a
/// third arm ([`Query`](BlobClientIntent::Query)) with no analogue in either
/// of those, so it defines its own enum rather than reusing that shared type.
/// This protocol's client-initiated requests.
#[derive(Debug, Serialize, Deserialize)]
enum BlobClientIntent {
Fetch(BlobSyncRequest),
@@ -202,35 +197,13 @@ pub struct BlobSyncResult {
/// Register with `Router::accept(BLOB_SYNC_ALPN, handler)`.
#[derive(Clone)]
pub struct BlobSyncHandler {
store: Arc<Store>,
}
impl BlobSyncHandler {
pub fn new(store: Arc<Store>) -> Self {
Self { store }
}
}
impl std::fmt::Debug for BlobSyncHandler {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("BlobSyncHandler").field("store_path", &self.store.path).finish()
}
pub(crate) store: Arc<Store>,
}
impl EnsembleProtocol for BlobSyncHandler {
async fn serve(
&self,
conn: Connection,
peer: CertifiedPeer,
) -> std::result::Result<(), AcceptError> {
self.run(conn, peer)
.await
.map_err(|e| AcceptError::from_err(std::io::Error::other(e.to_string())))
}
}
type Error = BlobSyncError;
impl BlobSyncHandler {
async fn run(&self, conn: Connection, _peer: CertifiedPeer) -> Result<()> {
async fn serve(&self, conn: Connection, _peer: CertifiedPeer) -> Result<()> {
let remote = conn.remote_id();
let (mut send, mut recv) = wire::accept(&conn).await?;
@@ -245,7 +218,9 @@ impl BlobSyncHandler {
conn.closed().await;
Ok(())
}
}
impl BlobSyncHandler {
/// Answer a `Fetch`: send back every requested blob we hold.
async fn serve_fetch(
&self,
rust/lib/src/p2p/doc_sync.rs
+5 -32
View File
@@ -20,9 +20,7 @@
use std::collections::HashMap;
use std::sync::Arc;
use iroh::{
EndpointAddr, EndpointId, KeyParsingError, endpoint::Connection, protocol::AcceptError,
};
use iroh::{EndpointAddr, EndpointId, KeyParsingError, endpoint::Connection};
use log::info;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
@@ -601,39 +599,14 @@ pub enum SyncProgress {
/// Register with `Router::accept(DOC_SYNC_ALPN, handler)`.
#[derive(Clone)]
pub struct DocSyncHandler {
store: Arc<Store>,
local_node: NodePubkey,
}
impl DocSyncHandler {
pub fn new(store: Arc<Store>, local_node: NodePubkey) -> Self {
Self { store, local_node }
}
}
impl std::fmt::Debug for DocSyncHandler {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("DocSyncHandler")
.field("store_path", &self.store.path)
.field("local_node", &self.local_node)
.finish()
}
pub(crate) store: Arc<Store>,
pub(crate) local_node: NodePubkey,
}
impl EnsembleProtocol for DocSyncHandler {
async fn serve(
&self,
conn: Connection,
peer: CertifiedPeer,
) -> std::result::Result<(), AcceptError> {
self.run(conn, peer)
.await
.map_err(|e| AcceptError::from_err(std::io::Error::other(e.to_string())))
}
}
type Error = DocSyncError;
impl DocSyncHandler {
async fn run(&self, conn: Connection, peer: CertifiedPeer) -> Result<()> {
async fn serve(&self, conn: Connection, peer: CertifiedPeer) -> Result<()> {
let remote = conn.remote_id();
// Equal to `remote` by construction — the admission handshake checked
// it — but taken from the certified peer so the provenance recorded
rust/lib/src/p2p/mod.rs
+39 -96
View File
@@ -32,23 +32,13 @@ pub(crate) mod wire_time;
use std::time::{Duration, Instant};
/// The first message on a client-initiated value-transfer session: whether
/// the connecting node wants to fetch data from the responder, or push data
/// to it. Shared by [`blob_sync`] and [`artifact_sync`], which each supply
/// their own protocol-specific fetch/push payload types.
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub(crate) enum ClientIntent<F, P> {
Fetch(F),
Push(P),
}
#[cfg(target_os = "android")]
use iroh::dns::{DnsProtocol, DnsResolver};
use iroh::{
Endpoint, EndpointAddr, EndpointId, RelayUrl,
address_lookup::memory::MemoryLookup,
endpoint::{Connection, presets::N0},
protocol::{AcceptError, Router},
protocol::Router,
tls::CaTlsConfig,
};
use iroh_mainline_address_lookup::DhtAddressLookup;
@@ -63,13 +53,15 @@ use crate::ensemble::secret::{Keypair, NodeSecret};
use crate::jobs::runner::JobRunner;
use crate::node::NodePubkey;
use crate::stores::{Store, peer_nodes::PeerNodeStore};
use admission::{CertifiedPeer, EnsembleEndpoint, EnsembleProtocol, Retiring};
use artifact_sync::ArtifactSyncHandler;
use blob_stream::BlobStreamHandler;
use blob_sync::BlobSyncHandler;
use doc_sync::DocSyncHandler;
use admission::{CertifiedPeer, EnsembleEndpoint, EnsembleProtocol, Gate, Retiring};
use artifact_sync::{ARTIFACT_SYNC_ALPN, ArtifactSyncHandler};
use blob_stream::{BLOB_STREAM_ALPN, BlobStreamHandler};
use blob_sync::{BLOB_SYNC_ALPN, BlobSyncHandler};
use doc_sync::{DOC_SYNC_ALPN, DocSyncHandler};
use gossip::{GossipDiscovery, GossipError};
use node_admin::{JobRunnerCell, NodeAdminHandler};
use node_admin::{JobRunnerCell, NODE_ADMIN_ALPN, NodeAdminHandler};
use revocation_delivery::{REVOCATION_ALPN, RevocationDeliveryHandler};
use wire::WireError;
/// Errors that can occur when starting a P2P node.
#[derive(Debug, thiserror::Error)]
@@ -136,28 +128,29 @@ pub const PING_ALPN: &[u8] = b"/syn/ping/2";
/// rather than only being refused by one (see `admission::AdmissionReply`).
pub const GATED_ALPNS: &[&[u8]] = &[
PING_ALPN,
revocation_delivery::REVOCATION_ALPN,
doc_sync::DOC_SYNC_ALPN,
artifact_sync::ARTIFACT_SYNC_ALPN,
blob_sync::BLOB_SYNC_ALPN,
blob_stream::BLOB_STREAM_ALPN,
node_admin::NODE_ADMIN_ALPN,
REVOCATION_ALPN,
DOC_SYNC_ALPN,
ARTIFACT_SYNC_ALPN,
BLOB_SYNC_ALPN,
BLOB_STREAM_ALPN,
NODE_ADMIN_ALPN,
];
#[derive(Debug, Clone)]
struct PingHandler;
impl EnsembleProtocol for PingHandler {
async fn serve(&self, conn: Connection, _peer: CertifiedPeer) -> Result<(), AcceptError> {
type Error = WireError;
async fn serve(&self, conn: Connection, _peer: CertifiedPeer) -> Result<(), WireError> {
let remote = conn.remote_id();
let (mut send, mut recv) = conn.accept_bi().await?;
let (mut send, mut recv) = wire::accept(&conn).await?;
let mut buf = [0u8; 4];
recv.read_exact(&mut buf).await.map_err(AcceptError::from_err)?;
recv.read_exact(&mut buf).await?;
if &buf == b"PING" {
info!("Received PING from {}", remote.fmt_short());
send.write_all(b"PONG").await.map_err(AcceptError::from_err)?;
send.write_all(b"PONG").await?;
send.finish()?;
}
@@ -284,85 +277,35 @@ impl P2PNode {
gossip_discovery.set_peer_health_store(store.peer_health_store.clone());
let local_node = NodePubkey(*keypair.secret_key().public().as_bytes());
let store_for_handlers = Arc::clone(&store);
let job_runner: JobRunnerCell = Arc::new(RwLock::new(None));
let retiring = Retiring::new();
let retiring = Retiring::default();
// Every application protocol is wrapped in `Gated`, which runs the
// delegation-cert handshake before the handler sees the connection.
// This is not a convention that can be forgotten: the handlers
// implement `EnsembleProtocol`, not iroh's `ProtocolHandler`, so
// registering one un-wrapped does not compile.
// A generic fn, not a closure: a closure would monomorphise to the
// first handler type it is called with.
fn gate<H: EnsembleProtocol>(
h: H,
store: &Arc<Store>,
retiring: &Retiring,
) -> admission::Gated<H> {
admission::Gated::new(h, Arc::clone(store), retiring.clone())
}
let gates = Gate { store: store.clone(), retiring: retiring.clone() };
let router = Router::builder(endpoint.clone())
.accept(iroh_gossip::ALPN, gossip_handle)
.accept(PING_ALPN, gate(PingHandler, &store_for_handlers, &retiring))
.accept(PING_ALPN, gates.gate(PingHandler))
.accept(REVOCATION_ALPN, gates.gate(RevocationDeliveryHandler { store: store.clone() }))
.accept(DOC_SYNC_ALPN, gates.gate(DocSyncHandler { store: store.clone(), local_node }))
.accept(ARTIFACT_SYNC_ALPN, gates.gate(ArtifactSyncHandler { store: store.clone() }))
.accept(BLOB_SYNC_ALPN, gates.gate(BlobSyncHandler { store: store.clone() }))
.accept(BLOB_STREAM_ALPN, gates.gate(BlobStreamHandler { store: store.clone() }))
.accept(
revocation_delivery::REVOCATION_ALPN,
gate(
revocation_delivery::RevocationDeliveryHandler::new(Arc::clone(
&store_for_handlers,
)),
&store_for_handlers,
&retiring,
),
)
.accept(
doc_sync::DOC_SYNC_ALPN,
gate(
DocSyncHandler::new(Arc::clone(&store_for_handlers), local_node),
&store_for_handlers,
&retiring,
),
)
.accept(
artifact_sync::ARTIFACT_SYNC_ALPN,
gate(
ArtifactSyncHandler::new(Arc::clone(&store_for_handlers)),
&store_for_handlers,
&retiring,
),
)
.accept(
blob_sync::BLOB_SYNC_ALPN,
gate(
BlobSyncHandler::new(Arc::clone(&store_for_handlers)),
&store_for_handlers,
&retiring,
),
)
.accept(
blob_stream::BLOB_STREAM_ALPN,
gate(
BlobStreamHandler::new(Arc::clone(&store_for_handlers)),
&store_for_handlers,
&retiring,
),
)
.accept(
node_admin::NODE_ADMIN_ALPN,
gate(
NodeAdminHandler::new(
Arc::clone(&store_for_handlers),
EnsembleEndpoint::new(
endpoint.clone(),
Arc::clone(&store_for_handlers),
retiring.clone(),
),
local_node,
Arc::clone(&job_runner),
NODE_ADMIN_ALPN,
gates.gate(NodeAdminHandler {
store: store.clone(),
endpoint: EnsembleEndpoint::new(
endpoint.clone(),
store.clone(),
retiring.clone(),
),
&store_for_handlers,
&retiring,
),
local_node,
job_runner: Arc::clone(&job_runner),
}),
)
.spawn();
rust/lib/src/p2p/node_admin.rs
+9 -36
View File
@@ -42,9 +42,7 @@
use std::collections::BTreeMap;
use std::sync::{Arc, RwLock};
use iroh::{
EndpointAddr, EndpointId, KeyParsingError, endpoint::Connection, protocol::AcceptError,
};
use iroh::{EndpointAddr, EndpointId, KeyParsingError, endpoint::Connection};
use log::info;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
@@ -188,43 +186,16 @@ pub type JobRunnerCell = Arc<RwLock<Option<Arc<JobRunner>>>>;
/// Register with `Router::accept(NODE_ADMIN_ALPN, handler)`.
#[derive(Clone)]
pub struct NodeAdminHandler {
store: Arc<Store>,
endpoint: EnsembleEndpoint,
local_node: NodePubkey,
job_runner: JobRunnerCell,
}
impl NodeAdminHandler {
pub fn new(
store: Arc<Store>,
endpoint: EnsembleEndpoint,
local_node: NodePubkey,
job_runner: JobRunnerCell,
) -> Self {
Self { store, endpoint, local_node, job_runner }
}
}
impl std::fmt::Debug for NodeAdminHandler {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("NodeAdminHandler").field("store_path", &self.store.path).finish()
}
pub(crate) store: Arc<Store>,
pub(crate) endpoint: EnsembleEndpoint,
pub(crate) local_node: NodePubkey,
pub(crate) job_runner: JobRunnerCell,
}
impl EnsembleProtocol for NodeAdminHandler {
async fn serve(
&self,
conn: Connection,
_peer: CertifiedPeer,
) -> std::result::Result<(), AcceptError> {
self.run(conn)
.await
.map_err(|e| AcceptError::from_err(std::io::Error::other(e.to_string())))
}
}
type Error = NodeAdminError;
impl NodeAdminHandler {
async fn run(&self, conn: Connection) -> Result<()> {
async fn serve(&self, conn: Connection, _peer: CertifiedPeer) -> Result<()> {
let remote = conn.remote_id();
let (mut send, mut recv) = wire::accept(&conn).await?;
@@ -237,7 +208,9 @@ impl NodeAdminHandler {
conn.closed().await;
Ok(())
}
}
impl NodeAdminHandler {
async fn handle(&self, request: AdminRequest) -> AdminResponse {
match request {
AdminRequest::GetSyncConfig => sync_config_response(&self.store),
rust/lib/src/p2p/revocation_delivery.rs
+4 -26
View File
@@ -43,7 +43,7 @@
use std::sync::Arc;
use iroh::{EndpointAddr, KeyParsingError, endpoint::Connection, protocol::AcceptError};
use iroh::{EndpointAddr, KeyParsingError, endpoint::Connection};
use log::{info, warn};
use serde::{Deserialize, Serialize};
@@ -118,37 +118,15 @@ pub struct RevocationsAccepted {
/// other protocol — see the module docs on why its gate is the permissive one.
#[derive(Clone)]
pub struct RevocationDeliveryHandler {
store: Arc<Store>,
}
impl RevocationDeliveryHandler {
pub fn new(store: Arc<Store>) -> Self {
Self { store }
}
}
impl std::fmt::Debug for RevocationDeliveryHandler {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("RevocationDeliveryHandler").field("store_path", &self.store.path).finish()
}
pub(crate) store: Arc<Store>,
}
impl EnsembleProtocol for RevocationDeliveryHandler {
const REVOKED_PEERS: RevokedPeers = RevokedPeers::Admit;
async fn serve(
&self,
conn: Connection,
_peer: CertifiedPeer,
) -> std::result::Result<(), AcceptError> {
self.run(conn)
.await
.map_err(|e| AcceptError::from_err(std::io::Error::other(e.to_string())))
}
}
type Error = RevocationDeliveryError;
impl RevocationDeliveryHandler {
async fn run(&self, conn: Connection) -> Result<()> {
async fn serve(&self, conn: Connection, _peer: CertifiedPeer) -> Result<()> {
let remote = conn.remote_id();
let (mut send, mut recv) = wire::accept(&conn).await?;
scripts/lint-p2p-admission.py
+3 -3
View File
@@ -19,7 +19,7 @@ what this lint closes:
1. `impl ProtocolHandler for ...` anywhere under `p2p/` except `admission.rs`
is an un-gated protocol. `Gated<H>` is the only intended implementor.
2. A `Router::builder(..).accept(..)` registration whose handler argument
does not go through `gate(`/`Gated::new(`. `iroh_gossip::ALPN` is exempt:
does not go through `gate(`. `iroh_gossip::ALPN` is exempt:
gossip is joinable by anyone who knows the ensemble public key and
authenticates per *message* instead, by verifying the delegation cert
carried in every announcement and purge intent.
@@ -95,7 +95,7 @@ def check_file(path: Path) -> list[str]:
findings.append(
f"{rel}:{n}: `impl ProtocolHandler for {m.group(1)}` — this protocol is "
f"served without the admission handshake. Implement `EnsembleProtocol` "
f"instead and register it via `Gated::new(..)`."
f"instead and register it via `Gate::gate(..)`."
)
if raw_endpoint_exempt(lines, n - 1):
@@ -134,7 +134,7 @@ def check_file(path: Path) -> list[str]:
alpn, handler = m.group(1).strip(), m.group(2)
if GOSSIP_EXEMPT in alpn:
continue
if "gate(" not in handler and "Gated::new(" not in handler:
if "gate(" not in handler:
findings.append(
f"{rel}:{n}: `.accept({alpn}, ..)` registers a handler that is not "
f"wrapped in `Gated`. Every ALPN but gossip must be gated."