Skip to content

Commit 4a2ceb9

Browse files
mickvandijkemickvandijke
committed
fix(relay): finalize provisional relay lifecycle
Carry the authoritative accepted connection into LinkTransport, keep provisional and published relay allocations behind one identity-scoped lifecycle, and remove packet-by-packet relay tracing from the hot path.\n\nCompletes WithAutonomi/saorsa-core#138 and supports WithAutonomi/saorsa-core#136.\n\nSemVer: breaking
1 parent 0830b24 commit 4a2ceb9

6 files changed

Lines changed: 288 additions & 356 deletions

File tree

‎src/link_transport_impl.rs‎

Lines changed: 56 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -650,37 +650,20 @@ impl LinkTransport for P2pLinkTransport {
650650
endpoint,
651651
|endpoint| async move {
652652
// Wait for an incoming connection
653-
if let Some(peer_conn) = endpoint.accept().await {
653+
if let Some((peer_conn, conn)) = endpoint.accept_with_connection().await {
654654
// Extract SocketAddr from TransportAddr
655655
let socket_addr = peer_conn
656656
.remote_addr
657657
.as_socket_addr()
658658
.unwrap_or_else(|| SocketAddr::from(([0, 0, 0, 0], 0)));
659659

660-
// Get the underlying QUIC connection by address
661-
match endpoint.get_quic_connection(&socket_addr).await {
662-
Ok(Some(conn)) => {
663-
// rustls exposes the authenticated identity as a
664-
// certificate vector whose first entry is the RFC
665-
// 7250 ML-DSA SPKI, not as a bare `Vec<u8>`.
666-
let public_key =
667-
crate::p2p_endpoint::extract_public_key_bytes_from_connection(
668-
&conn,
669-
);
670-
let link_conn = P2pLinkConn::new(conn, public_key, socket_addr);
671-
Some((Ok(link_conn), endpoint))
672-
}
673-
Ok(None) => {
674-
// Connection not found, try again
675-
Some((
676-
Err(LinkError::ConnectionFailed(
677-
"Connection not found".to_string(),
678-
)),
679-
endpoint,
680-
))
681-
}
682-
Err(e) => Some((Err(LinkError::ConnectionFailed(e.to_string())), endpoint)),
683-
}
660+
// rustls exposes the authenticated identity as a
661+
// certificate vector whose first entry is the RFC 7250
662+
// ML-DSA SPKI, not as a bare `Vec<u8>`.
663+
let public_key =
664+
crate::p2p_endpoint::extract_public_key_bytes_from_connection(&conn);
665+
let link_conn = P2pLinkConn::new(conn, public_key, socket_addr);
666+
Some((Ok(link_conn), endpoint))
684667
} else {
685668
// Endpoint is shutting down
686669
None
@@ -1440,6 +1423,54 @@ mod tests {
14401423
assert!(state.capabilities.is_empty());
14411424
}
14421425

1426+
#[tokio::test]
1427+
async fn accept_uses_the_authoritative_connection_handle() {
1428+
let bind_addr: SocketAddr = "127.0.0.1:0".parse().expect("valid bind address");
1429+
let server_endpoint = Arc::new(
1430+
P2pEndpoint::new(
1431+
P2pConfig::builder()
1432+
.bind_addr(bind_addr)
1433+
.build()
1434+
.expect("valid server config"),
1435+
)
1436+
.await
1437+
.expect("server endpoint"),
1438+
);
1439+
let server_addr = server_endpoint.local_addr().expect("server address");
1440+
let server = P2pLinkTransport::from_endpoint(Arc::clone(&server_endpoint));
1441+
let client = P2pEndpoint::new(
1442+
P2pConfig::builder()
1443+
.bind_addr(bind_addr)
1444+
.build()
1445+
.expect("valid client config"),
1446+
)
1447+
.await
1448+
.expect("client endpoint");
1449+
1450+
let mut incoming = server.accept(ProtocolId::DEFAULT);
1451+
client
1452+
.connect(server_addr)
1453+
.await
1454+
.expect("client connection");
1455+
let accepted = tokio::time::timeout(std::time::Duration::from_secs(10), incoming.next())
1456+
.await
1457+
.expect("accept timed out")
1458+
.expect("accept stream ended")
1459+
.expect("accepted connection");
1460+
1461+
assert_eq!(
1462+
accepted.remote_addr().ip(),
1463+
client.local_addr().unwrap().ip()
1464+
);
1465+
assert!(
1466+
accepted.peer_public_key().is_some(),
1467+
"accepted handle must retain its authenticated identity"
1468+
);
1469+
1470+
client.shutdown().await;
1471+
server_endpoint.shutdown().await;
1472+
}
1473+
14431474
// =========================================================================
14441475
// Phase 3: SharedTransport Tests
14451476
// =========================================================================

‎src/masque/mod.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -127,4 +127,5 @@ pub use relay_server::{
127127
pub use relay_session::{
128128
RelayPeerId, RelaySession, RelaySessionConfig, RelaySessionState, RelaySessionStats,
129129
};
130-
pub use relay_socket::{MasqueRelaySocket, RawRelayStreams, RelayTunnelControl};
130+
pub(crate) use relay_socket::RelayTunnelControl;
131+
pub use relay_socket::{MasqueRelaySocket, RawRelayStreams};

‎src/masque/relay_server.rs‎

Lines changed: 2 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -1379,12 +1379,6 @@ impl MasqueRelayServer {
13791379
match socket.recv_from(&mut buf).await {
13801380
Ok((len, source)) => {
13811381
let payload = Bytes::copy_from_slice(&buf[..len]);
1382-
tracing::trace!(
1383-
session_id,
1384-
source = %source,
1385-
len,
1386-
"RELAY_TUNNEL[srv]: dgram-loop dir1 recv UDP → forwarding to relay-client"
1387-
);
13881382

13891383
// Encode as uncompressed datagram (includes source address
13901384
// so client can decode without context registration)
@@ -1452,24 +1446,10 @@ impl MasqueRelayServer {
14521446
};
14531447
match resolved {
14541448
Some((target, payload)) => {
1455-
tracing::trace!(
1456-
session_id,
1457-
target = %target,
1458-
len = payload.len(),
1459-
"RELAY_TUNNEL[srv]: dgram-loop dir2 recv from relay-client → sendto target"
1460-
);
14611449
server2.stats.record_bytes(payload.len() as u64);
14621450
server2.stats.record_datagram();
14631451
match socket2.send_to(&payload, target).await {
1464-
Ok(n) => {
1465-
tracing::trace!(
1466-
session_id,
1467-
target = %target,
1468-
len = payload.len(),
1469-
sent = n,
1470-
"RELAY_TUNNEL[srv]: dgram-loop dir2 sendto OK"
1471-
);
1472-
}
1452+
Ok(_) => {}
14731453
Err(e) => {
14741454
tracing::warn!(
14751455
session_id,
@@ -1611,10 +1591,6 @@ impl MasqueRelayServer {
16111591
match socket.recv_from(&mut buf).await {
16121592
Ok((len, source)) => {
16131593
let payload = Bytes::copy_from_slice(&buf[..len]);
1614-
tracing::trace!(
1615-
session_id, source = %source, len,
1616-
"RELAY_TUNNEL[srv]: stream-loop dir1 recv UDP → forwarding to relay-client"
1617-
);
16181594
let datagram =
16191595
UncompressedDatagram::new(VarInt::from_u32(0), source, payload);
16201596
let encoded = datagram.encode();
@@ -1775,26 +1751,14 @@ impl MasqueRelayServer {
17751751
let mut cursor = Bytes::from(frame_buf);
17761752
match UncompressedDatagram::decode(&mut cursor) {
17771753
Ok(datagram) => {
1778-
tracing::trace!(
1779-
session_id, target = %datagram.target,
1780-
len = datagram.payload.len(),
1781-
"RELAY_TUNNEL[srv]: stream-loop dir2 recv from relay-client → sendto target"
1782-
);
17831754
stats2.record_bytes(datagram.payload.len() as u64);
17841755
stats2.record_datagram();
17851756
let target = datagram.target;
17861757
let payload_len = datagram.payload.len();
17871758
match socket2.send_to(&datagram.payload, target).await {
1788-
Ok(n) => {
1759+
Ok(_) => {
17891760
// Confirmed forwarded to the third-party target.
17901761
stats2.record_forwarded_to_target(payload_len as u64, 1);
1791-
tracing::trace!(
1792-
session_id,
1793-
target = %target,
1794-
len = payload_len,
1795-
sent = n,
1796-
"RELAY_TUNNEL[srv]: stream-loop dir2 sendto OK"
1797-
);
17981762
}
17991763
Err(e) if is_message_too_large(&e) => {
18001764
// Path-MTU exceeded. Emit a PmtuUpdate

‎src/masque/relay_socket.rs‎

Lines changed: 5 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -108,7 +108,7 @@ pub struct RawRelayStreams {
108108
/// promptly tells the relay server to close the associated MASQUE session and
109109
/// release its capacity slot. Shutdown is idempotent.
110110
#[derive(Debug)]
111-
pub struct RelayTunnelControl {
111+
pub(crate) struct RelayTunnelControl {
112112
tasks: PlMutex<Vec<tokio::task::JoinHandle<()>>>,
113113
closed: Notify,
114114
is_closed: AtomicBool,
@@ -144,12 +144,12 @@ impl RelayTunnelControl {
144144
}
145145

146146
/// Returns whether the tunnel has failed or has been explicitly shut down.
147-
pub fn is_closed(&self) -> bool {
147+
pub(crate) fn is_closed(&self) -> bool {
148148
self.is_closed.load(Ordering::Acquire)
149149
}
150150

151151
/// Wait until the tunnel reader exits or shutdown is requested.
152-
pub async fn closed(&self) {
152+
pub(crate) async fn closed(&self) {
153153
loop {
154154
if self.is_closed() {
155155
return;
@@ -163,7 +163,7 @@ impl RelayTunnelControl {
163163
}
164164

165165
/// Stop the tunnel and wait for all task-owned QUIC streams to be dropped.
166-
pub async fn shutdown(&self) {
166+
pub(crate) async fn shutdown(&self) {
167167
self.mark_closed();
168168
let handles = {
169169
let mut tasks = self.tasks.lock();
@@ -252,7 +252,7 @@ impl MasqueRelaySocket {
252252
/// Without this, the driver's `Drop` impl fires last and cascades
253253
/// a cryptic `"endpoint driver future was dropped"` into every
254254
/// connection accepted through this tunnel.
255-
pub fn new(
255+
pub(crate) fn new(
256256
mut send_stream: crate::high_level::SendStream,
257257
mut recv_stream: crate::high_level::RecvStream,
258258
relay_public_addr: SocketAddr,
@@ -358,14 +358,6 @@ impl MasqueRelaySocket {
358358
Ok(datagram) => {
359359
// `datagram.payload` is a zero-copy slice of
360360
// the original frame buffer — no clone needed.
361-
let inbound_source = datagram.target;
362-
let inbound_len = datagram.payload.len();
363-
tracing::trace!(
364-
relay = %relay_public_addr,
365-
source = %inbound_source,
366-
len = inbound_len,
367-
"RELAY_TUNNEL[clt]: decoded inbound frame → enqueue for Quinn poll_recv"
368-
);
369361
if recv_tx
370362
.send((datagram.payload, datagram.target))
371363
.await
@@ -507,14 +499,6 @@ impl AsyncUdpSocket for MasqueRelaySocket {
507499
// of `segment_size` bytes. Each segment must be sent as its
508500
// own tunnel frame — the relay server has a per-frame size
509501
// limit and cannot handle the entire batch as one.
510-
tracing::trace!(
511-
relay = %self.relay_public_addr,
512-
destination = %transmit.destination,
513-
len = transmit.contents.len(),
514-
segment_size = ?transmit.segment_size,
515-
"RELAY_TUNNEL[clt]: try_send → enqueue outbound for relay-server"
516-
);
517-
518502
// Per-target MTU enforcement: if a previous PmtuUpdate control
519503
// frame told us the egress path to this destination caps at
520504
// `mtu` bytes, drop oversized packets here so they never reach
@@ -599,12 +583,6 @@ impl AsyncUdpSocket for MasqueRelaySocket {
599583
recv_meta.dst_ip = None;
600584
meta[filled] = recv_meta;
601585

602-
tracing::trace!(
603-
source = %source,
604-
len,
605-
"RELAY_TUNNEL: recv from tunnel queue"
606-
);
607-
608586
filled += 1;
609587
}
610588
Poll::Ready(None) => {

0 commit comments

Comments
 (0)