Implement pairing link lifecycle management

Author: yosebyte

docs/protocol.md

@@ -337,6 +337,12 @@ without an application response. Network closure is initiated before detailed
 failure information is written to the Portal's local log. Service shutdown
 cancels the delay.
 
+If a newly authenticated QUIC connection presents the `session_id` of an
+older active QUIC connection, the Portal replaces the older carrier. This
+permits recovery from a client-side path failure while the stale server-side
+connection is still waiting for its idle timeout. Existing flows on the older
+carrier are not migrated.
+
 The Portal applies a process-wide pre-authentication admission limit shared by
 TCP and QUIC: at most 256 connections in total and 32 per IPv4 `/32` or IPv6
 `/64`. A validated QUIC attempt above either limit is silently ignored; an
@@ -429,7 +435,10 @@ Every frame begins with `version_u8(1) || type_u8`. Before receiving
 `OPEN_ACK`, every client payload uses `OPEN_DATA`; after the ACK, the client
 uses target-free `DATA`. The Portal treats repeated open metadata idempotently
 and forwards the first payload without an additional RTT. Target-to-client
-DATAGRAMs always use compact `DATA`.
+DATAGRAMs always use compact `DATA`. If the Portal receives target-free `DATA`
+for an unknown or expired flow, it returns `CLOSE`; the client then clears its
+ACK state and uses `OPEN_DATA` for the next payload, or recreates an asymmetric
+flow, to restore the target metadata.
 
 The derived-order header below remains accepted for the existing symmetric
 implementation path but is not emitted by the upgraded reference client.

src/portal/conn.rs

@@ -13,9 +13,6 @@ use std::sync::Arc;
 use quinn::{Connection, Incoming, VarInt};
 use tokio_util::sync::CancellationToken;
 
-use crate::common::{quic_max_streams, rate_limit_bytes_per_second};
-use crate::protocol::Carrier;
-
 use self::auth::{
     AuthenticationOutcome, authenticate_connection, authentication_deadline,
     authentication_failure_close,
@@ -34,6 +31,7 @@ use super::admission::UnauthenticatedGuard;
 use super::admission::{
     MAX_UNAUTHENTICATED_CONNECTIONS, MAX_UNAUTHENTICATED_PER_SOURCE, UnauthenticatedAdmission,
 };
+use crate::common::{quic_max_streams, rate_limit_bytes_per_second};
 
 pub(super) async fn handle_incoming(
     portal: Arc<PortalInner>,
@@ -82,20 +80,14 @@ async fn handle_connection(
     conn.set_max_concurrent_bi_streams(VarInt::from_u32(quic_max_streams()));
     drop(admission);
     let session = authenticated.session;
-    let _link_guard =
-        match portal
-            .pairing
-            .register_link(session.session_id, Carrier::Udp, portal.stats.clone())
-        {
-            Ok(guard) => guard,
-            Err(err) => {
-                conn.close(VarInt::from_u32(1), b"duplicate session");
-                portal.logger.error(format_args!(
-                    "portal::conn::handle_connection: failed to register link: {err}"
-                ));
-                return;
-            }
-        };
+    let link_replaced = CancellationToken::new();
+    let link_guard = portal.pairing.register_quic_link(
+        session.session_id,
+        portal.stats.clone(),
+        link_replaced.clone(),
+    );
+    session.set_quic_generation(link_guard.quic_generation());
+    let _link_guard = link_guard;
 
     let etar_bps = rate_limit_bytes_per_second(portal.etar_limit);
     if etar_bps > 0 {
@@ -113,6 +105,12 @@ async fn handle_connection(
     loop {
         tokio::select! {
             _ = shutdown.cancelled() => break,
+            _ = link_replaced.cancelled() => {
+                portal.logger.debug(format_args!(
+                    "portal::conn::handle_connection: authenticated QUIC carrier replaced"
+                ));
+                break;
+            },
             stream = conn.accept_bi() => {
                 match stream {
                     Ok((send, recv)) => {

src/portal/conn/relay_uot.rs

@@ -13,8 +13,8 @@ use crate::common::{handshake_timeout, udp_dial_timeout, udp_idle_timeout};
 use crate::portal::PortalInner;
 use crate::portal::pairing::PairedUdp;
 use crate::protocol::{
-    Carrier, DATAGRAM_UDP_CLOSE, DATAGRAM_UDP_DATA, DATAGRAM_UDP_OPEN_ACK, encode_udp_compact,
-    read_uot_packet, read_uot_setup_target, write_uot_packet_frame,
+    Carrier, DATAGRAM_UDP_COMPACT_CLOSE, DATAGRAM_UDP_DATA, DATAGRAM_UDP_OPEN_ACK,
+    encode_udp_compact, read_uot_packet, read_uot_setup_target, write_uot_packet_frame,
 };
 
 use super::{SessionGuard, paired_exchange_path, symmetric_exchange_path};
@@ -159,6 +159,7 @@ pub(in crate::portal) async fn relay_paired_udp(portal: Arc<PortalInner>, paired
         downlink_carrier,
         uplink_path,
         downlink_path,
+        compact_ack,
     } = paired;
     let socket = match portal
         .outbound
@@ -190,12 +191,7 @@ pub(in crate::portal) async fn relay_paired_udp(portal: Arc<PortalInner>, paired
     ));
     portal.stats.add_session(true);
     let _done = SessionGuard::new(portal.clone(), true);
-    if let Err(err) = send_udp_ack(&mut downlink, flow_id).await {
-        portal.logger.debug(format_args!(
-            "portal::conn::relay_paired_udp: failed to send ACK: {err}"
-        ));
-        return;
-    }
+    let mut ack_sent = false;
     let mut target_buf = portal.buffers.get_udp_buffer();
     let mut last_used = Instant::now();
     loop {
@@ -217,6 +213,17 @@ pub(in crate::portal) async fn relay_paired_udp(portal: Arc<PortalInner>, paired
                 }
                 match socket.send(&payload).await {
                     Ok(n) => {
+                        if !ack_sent {
+                            if let Err(err) =
+                                send_udp_ack(&mut downlink, flow_id, compact_ack.as_ref()).await
+                            {
+                                portal.logger.debug(format_args!(
+                                    "portal::conn::relay_paired_udp: failed to send ACK: {err}"
+                                ));
+                                break;
+                            }
+                            ack_sent = true;
+                        }
                         portal.stats.udp_rx.fetch_add(n as u64, Ordering::Relaxed);
                         match uplink_carrier {
                             Carrier::Tcp => &portal.stats.up_tcp,
@@ -269,16 +276,24 @@ async fn read_paired_udp(
 async fn send_udp_ack(
     downlink: &mut crate::portal::pairing::UdpDown,
     flow_id: u64,
+    compact_ack: Option<&crate::portal::pairing::CompactAck>,
 ) -> anyhow::Result<()> {
     match downlink {
         crate::portal::pairing::UdpDown::Tcp(writer) => {
             writer.write_all(&[0, 0]).await?;
         }
         crate::portal::pairing::UdpDown::Quic(conn) => {
-            let frame = encode_udp_compact(DATAGRAM_UDP_OPEN_ACK, flow_id, &[])?;
-            conn.send_datagram(bytes::Bytes::from(frame))?;
+            if compact_ack.is_none() {
+                let frame = encode_udp_compact(DATAGRAM_UDP_OPEN_ACK, flow_id, &[])?;
+                conn.send_datagram(bytes::Bytes::from(frame))?;
+            }
         }
     }
+    if let Some(ack) = compact_ack {
+        let frame = encode_udp_compact(DATAGRAM_UDP_OPEN_ACK, flow_id, &[])?;
+        ack.conn.send_datagram(bytes::Bytes::from(frame))?;
+        ack.acked.store(true, Ordering::Release);
+    }
     Ok(())
 }
 
@@ -308,7 +323,7 @@ async fn send_udp_close(
             writer.write_all(&[0, 0]).await?;
         }
         crate::portal::pairing::UdpDown::Quic(conn) => {
-            let frame = encode_udp_compact(DATAGRAM_UDP_CLOSE, flow_id, &[])?;
+            let frame = encode_udp_compact(DATAGRAM_UDP_COMPACT_CLOSE, flow_id, &[])?;
             conn.send_datagram(bytes::Bytes::from(frame))?;
         }
     }

src/portal/conn/session.rs

@@ -10,7 +10,7 @@ mod flow;
 
 use std::collections::HashMap;
 use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 
 use bytes::Bytes;
 use quinn::{Connection, RecvStream, SendStream};
@@ -35,6 +35,7 @@ pub(super) struct PortalSession {
     portal: Arc<PortalInner>,
     conn: Connection,
     pub(super) session_id: SessionId,
+    quic_generation: AtomicU64,
     udp_flows: Mutex<HashMap<UdpFlowKey, Arc<PortalUdpFlow>>>,
     compact_udp_flows: Mutex<HashMap<u64, CompactUdpState>>,
     udp_queue_budget: Arc<Semaphore>,
@@ -46,6 +47,7 @@ pub(super) struct CompactUdpState {
     pub(super) target: String,
     pub(super) downlink: Carrier,
     pub(super) sender: mpsc::Sender<Bytes>,
+    pub(super) acked: Arc<AtomicBool>,
 }
 
 impl PortalSession {
@@ -66,6 +68,7 @@ impl PortalSession {
             portal,
             conn,
             session_id,
+            quic_generation: AtomicU64::new(0),
             udp_flows: Mutex::new(HashMap::new()),
             compact_udp_flows: Mutex::new(HashMap::new()),
             udp_queue_budget,
@@ -74,6 +77,14 @@ impl PortalSession {
         }
     }
 
+    pub(super) fn set_quic_generation(&self, generation: u64) {
+        self.quic_generation.store(generation, Ordering::Release);
+    }
+
+    pub(super) fn quic_generation(&self) -> u64 {
+        self.quic_generation.load(Ordering::Acquire)
+    }
+
     /// Handles a bidirectional QUIC stream carrying one TCP target request.
     pub(super) async fn handle_stream(self: Arc<Self>, mut send: SendStream, recv: RecvStream) {
         let portal = &self.portal;
@@ -123,7 +134,10 @@ impl PortalSession {
                                     self.session_id,
                                     header,
                                     target_addr,
-                                    self.link_path(),
+                                    crate::portal::pairing::LinkHalf::quic(
+                                        self.link_path(),
+                                        self.quic_generation(),
+                                    ),
                                     Some(Box::pin(recv)),
                                     None,
                                 )
@@ -136,7 +150,10 @@ impl PortalSession {
                                     self.session_id,
                                     header,
                                     target_addr,
-                                    self.link_path(),
+                                    crate::portal::pairing::LinkHalf::quic(
+                                        self.link_path(),
+                                        self.quic_generation(),
+                                    ),
                                     None,
                                     Some(Box::pin(send)),
                                 )
@@ -166,9 +183,13 @@ impl PortalSession {
                             self.session_id,
                             header,
                             target_addr,
-                            self.link_path(),
-                            None,
-                            Some(crate::portal::pairing::UdpDown::Quic(self.conn.clone())),
+                            crate::portal::pairing::LinkHalf::quic(
+                                self.link_path(),
+                                self.quic_generation(),
+                            ),
+                            crate::portal::pairing::UdpHalf::Downlink(
+                                crate::portal::pairing::UdpDown::Quic(self.conn.clone()),
+                            ),
                         )
                         .await;
                     match result {

src/portal/conn/session_datagram.rs

@@ -5,16 +5,18 @@
 
 use std::collections::VecDeque;
 use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering;
 
 use bytes::Bytes;
 use tokio::sync::OwnedSemaphorePermit;
 use tokio_util::sync::CancellationToken;
 
 use crate::protocol::{
-    Carrier, CompactUdpFrame, DATAGRAM_UDP_CLOSE, DATAGRAM_UDP_REQUEST, DATAGRAM_UDP_RESPONSE,
-    FlowHeader, FlowKind, FlowRole, decode_udp_compact, decode_udp_datagram_parts,
-    encode_udp_compact, new_udp_datagram_header,
+    Carrier, CompactUdpFrame, DATAGRAM_UDP_CLOSE, DATAGRAM_UDP_COMPACT_CLOSE,
+    DATAGRAM_UDP_OPEN_ACK, DATAGRAM_UDP_REQUEST, DATAGRAM_UDP_RESPONSE, FlowHeader, FlowKind,
+    FlowRole, decode_udp_compact, decode_udp_datagram_parts, encode_udp_compact,
+    new_udp_datagram_header,
 };
 
 use super::flow::{PortalUdpFlow, QueuedDatagram, UdpFlowKey};
@@ -112,11 +114,18 @@ impl PortalSession {
                         (
                             state.target == target && state.downlink == downlink,
                             state.sender.clone(),
+                            state.acked.load(Ordering::Acquire),
                         )
                     });
-                if let Some((valid, sender)) = existing {
+                if let Some((valid, sender, acked)) = existing {
                     if valid {
                         let _ = sender.try_send(payload);
+                        if acked
+                            && let Ok(frame) =
+                                encode_udp_compact(DATAGRAM_UDP_OPEN_ACK, flow_id, &[])
+                        {
+                            let _ = self.conn.send_datagram(Bytes::from(frame));
+                        }
                     } else {
                         drop(sender);
                         self.reject_compact_flow(flow_id).await;
@@ -130,12 +139,14 @@ impl PortalSession {
                     return;
                 }
                 let (sender, receiver) = tokio::sync::mpsc::channel(64);
+                let acked = Arc::new(AtomicBool::new(false));
                 self.compact_udp_flows.lock().await.insert(
                     flow_id,
                     CompactUdpState {
                         target: target.clone(),
                         downlink,
                         sender: sender.clone(),
+                        acked: acked.clone(),
                     },
                 );
                 let _ = sender.try_send(payload);
@@ -159,6 +170,10 @@ impl PortalSession {
                         downlink_carrier: Carrier::Udp,
                         uplink_path: path.clone(),
                         downlink_path: path,
+                        compact_ack: Some(crate::portal::pairing::CompactAck {
+                            conn: self.conn.clone(),
+                            acked,
+                        }),
                     })
                 } else {
                     let header = FlowHeader {
@@ -175,9 +190,17 @@ impl PortalSession {
                             self.session_id,
                             header,
                             target,
-                            self.link_path(),
-                            Some(crate::portal::pairing::UdpUp::Quic(receiver)),
-                            None,
+                            crate::portal::pairing::LinkHalf::quic(
+                                self.link_path(),
+                                self.quic_generation(),
+                            ),
+                            crate::portal::pairing::UdpHalf::Uplink {
+                                uplink: crate::portal::pairing::UdpUp::Quic(receiver),
+                                compact_ack: Some(crate::portal::pairing::CompactAck {
+                                    conn: self.conn.clone(),
+                                    acked,
+                                }),
+                            },
                         )
                         .await
                     {
@@ -199,15 +222,17 @@ impl PortalSession {
                 }
             }
             CompactUdpFrame::Data { flow_id, payload } => {
-                if let Some(sender) = self
+                let sender = self
                     .compact_udp_flows
                     .lock()
                     .await
                     .get(&flow_id)
-                    .map(|s| s.sender.clone())
-                {
+                    .map(|s| s.sender.clone());
+                if let Some(sender) = sender {
                     let payload_offset = payload.as_ptr() as usize - data.as_ptr() as usize;
                     let _ = sender.try_send(data.slice(payload_offset..));
+                } else {
+                    self.reject_compact_flow(flow_id).await;
                 }
             }
             CompactUdpFrame::Close { flow_id } => {
@@ -227,7 +252,7 @@ impl PortalSession {
             .pairing
             .cancel_udp(self.session_id, flow_id)
             .await;
-        if let Ok(frame) = encode_udp_compact(DATAGRAM_UDP_CLOSE, flow_id, &[]) {
+        if let Ok(frame) = encode_udp_compact(DATAGRAM_UDP_COMPACT_CLOSE, flow_id, &[]) {
             let _ = self.conn.send_datagram(Bytes::from(frame));
         }
     }