fix(dataplane): finish TCP teardown after stream close (#2754)

Dropping a data-plane TCP stream removed its socket without sending FIN,
leaving peer reads and exit-node forwarding tasks waiting indefinitely.

Transfer released smoltcp sockets, flow registrations, and stack leases
to the existing reactor. Preserve FIN retransmissions and TIME-WAIT
after application handles close, drain unread peer data, and reclaim
sockets on completion, an absolute close deadline, or reactor shutdown.

Reuse session resource closure to cancel pending reads and writes. Keep
the Go host on the existing resource-close ABI and refresh its embedded
WASI guest, protobuf bindings, and matching provenance together.

Cover blocked writes, expired deadlines, lost FIN and ACK packets, close
deadlines, reactor shutdown, and final flow and stack lease release.
Validate with 77 Rust tests, the Go suite, fmt, and strict workspace
clippy.

Co-authored-by: KKRainbow <5665404+KKRainbow@users.noreply.github.com>

easytier-core/src/gateway/dataplane/mod.rs

@@ -176,7 +176,19 @@ const TCP_ENTRY: FlowKind = FlowKind::Tcp;
 const TCP_LISTEN_ENTRY: FlowKind = FlowKind::TcpListen;
 
 type FlowSet = Arc<FlowTable<FlowData>>;
-type DataPlaneTcpIo = Box<dyn TcpProxyStream>;
+pub(super) enum DataPlaneTcpIo {
+    Smoltcp(crate::gateway::smoltcp::TcpStream),
+    External(Box<dyn TcpProxyStream>),
+}
+
+impl DataPlaneTcpIo {
+    fn as_mut(&mut self) -> &mut dyn TcpProxyStream {
+        match self {
+            Self::Smoltcp(stream) => stream,
+            Self::External(stream) => &mut **stream,
+        }
+    }
+}
 
 pub(crate) struct DataPlaneRuntime<H>
 where
@@ -688,7 +700,7 @@ where
         generation.ensure_open()?;
 
         Ok(DataPlaneTcpStream::new(
-            Box::new(stream),
+            DataPlaneTcpIo::Smoltcp(stream),
             local_addr,
             Some(data_plane_ref),
             DataPlaneTcpStreamRoute::Outbound { _flow: flow },
@@ -710,7 +722,7 @@ where
             .await?;
         let local_addr = socket.local_addr().map_err(DataPlaneError::from)?;
         Ok(DataPlaneTcpStream::new(
-            Box::new(socket),
+            DataPlaneTcpIo::External(Box::new(socket)),
             local_addr,
             None,
             DataPlaneTcpStreamRoute::External,
@@ -748,7 +760,7 @@ where
             )))
             .await?;
         Ok(DataPlaneTcpStream::new(
-            stream,
+            DataPlaneTcpIo::External(stream),
             source_addr,
             None,
             DataPlaneTcpStreamRoute::External,

easytier-core/src/gateway/dataplane/tcp.rs

@@ -25,7 +25,7 @@ pub(super) enum DataPlaneTcpStreamRoute {
 
 /// A TCP stream created by the data-plane API.
 pub struct DataPlaneTcpStream {
-    pub(super) stream: DataPlaneTcpIo,
+    stream: Option<DataPlaneTcpIo>,
     pub(super) local_addr: SocketAddr,
     pub(super) _route: DataPlaneTcpStreamRoute,
     pub(super) _data_plane_lease: Option<DataPlaneLease>,
@@ -53,7 +53,7 @@ impl DataPlaneTcpStream {
         generation: DataPlaneIoGuard,
     ) -> Self {
         Self {
-            stream,
+            stream: Some(stream),
             local_addr,
             _data_plane_lease: data_plane_lease,
             _route: route,
@@ -68,6 +68,21 @@ impl DataPlaneTcpStream {
     }
 }
 
+impl Drop for DataPlaneTcpStream {
+    fn drop(&mut self) {
+        if let Some(DataPlaneTcpIo::Smoltcp(stream)) = self.stream.take() {
+            if self.generation.is_closed() {
+                return;
+            }
+            // The application is done, but FIN retransmissions and the peer's
+            // FIN still need this flow and the network stack until TCP closes.
+            let route = std::mem::replace(&mut self._route, DataPlaneTcpStreamRoute::External);
+            let lease = self._data_plane_lease.take();
+            stream.close((route, lease));
+        }
+    }
+}
+
 impl DataPlaneTcpListener {
     pub fn local_addr(&self) -> SocketAddr {
         self.local_addr
@@ -99,7 +114,7 @@ impl DataPlaneTcpListener {
             FlowData::DataPlaneRoute,
         );
         let accepted = DataPlaneTcpStream::new(
-            Box::new(stream),
+            DataPlaneTcpIo::Smoltcp(stream),
             local_addr,
             Some(self.data_plane_lease.clone()),
             DataPlaneTcpStreamRoute::Accepted { _flow: flow },
@@ -119,7 +134,7 @@ impl AsyncRead for DataPlaneTcpStream {
         if this.generation.ensure_open().is_err() || this.read_closed.as_mut().poll(cx).is_ready() {
             return Poll::Ready(Err(this.generation.closed_io_error()));
         }
-        Pin::new(&mut this.stream).poll_read(cx, buf)
+        Pin::new(this.stream.as_mut().unwrap().as_mut()).poll_read(cx, buf)
     }
 }
 
@@ -134,7 +149,7 @@ impl AsyncWrite for DataPlaneTcpStream {
         {
             return Poll::Ready(Err(this.generation.closed_io_error()));
         }
-        Pin::new(&mut this.stream).poll_write(cx, buf)
+        Pin::new(this.stream.as_mut().unwrap().as_mut()).poll_write(cx, buf)
     }
 
     fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), std::io::Error>> {
@@ -143,7 +158,7 @@ impl AsyncWrite for DataPlaneTcpStream {
         {
             return Poll::Ready(Err(this.generation.closed_io_error()));
         }
-        Pin::new(&mut this.stream).poll_flush(cx)
+        Pin::new(this.stream.as_mut().unwrap().as_mut()).poll_flush(cx)
     }
 
     fn poll_shutdown(
@@ -155,6 +170,6 @@ impl AsyncWrite for DataPlaneTcpStream {
         {
             return Poll::Ready(Err(this.generation.closed_io_error()));
         }
-        Pin::new(&mut this.stream).poll_shutdown(cx)
+        Pin::new(this.stream.as_mut().unwrap().as_mut()).poll_shutdown(cx)
     }
 }

easytier-core/src/gateway/dataplane/tests.rs

@@ -542,9 +542,29 @@ async fn listener_and_accepted_stream_own_independent_flow_lifetimes() {
     assert_eq!(&buf, b"after-listener-drop");
 
     drop(client);
-    assert_eq!(a.gateway.entries.count(), 0);
+    assert_eq!(a.gateway.entries.count(), 1);
+    assert_eq!(server.read(&mut buf).await.unwrap(), 0);
+    // The closed application no longer reads, but TCP must keep draining
+    // incoming data so the peer's queued data and FIN can finish sending.
+    tokio::time::timeout(timeout, async {
+        server.write_all(&vec![0; 1024 * 1024]).await.unwrap();
+        server.shutdown().await.unwrap();
+    })
+    .await
+    .unwrap();
     drop(server);
-    assert_eq!(b.gateway.entries.count(), 0);
+
+    tokio::time::timeout(Duration::from_secs(15), async {
+        while a.gateway.entries.count() != 0
+            || b.gateway.entries.count() != 0
+            || a.gateway.net.lock().await.is_some()
+            || b.gateway.net.lock().await.is_some()
+        {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("TCP teardown must release the last flow and stack leases");
 
     stop_data_plane_pair(&a, &b).await;
 }

easytier-core/src/gateway/smoltcp/mod.rs

@@ -3,5 +3,5 @@ mod tokio_smoltcp;
 
 pub(super) use stack::{SmolTcpStack, output_dst_ip};
 pub(super) use tokio_smoltcp::{
-    BufferSize, Net, NetConfig, TcpListener, UdpSocket, channel_device,
+    BufferSize, Net, NetConfig, TcpListener, TcpStream, UdpSocket, channel_device,
 };

easytier-core/src/gateway/smoltcp/tokio_smoltcp/reactor.rs

@@ -24,7 +24,10 @@ use std::{
     sync::Arc,
     task::{Context as TaskContext, Poll, Waker},
 };
-use tokio::{pin, select, sync::Notify};
+use tokio::{
+    pin, select,
+    sync::{Notify, mpsc},
+};
 
 use crate::foundation::time::sleep;
 
@@ -50,11 +53,18 @@ struct TcpListenerRegistry {
 
 type SharedTcpListenerRegistry = Arc<Mutex<TcpListenerRegistry>>;
 
+struct ClosingTcpSocket {
+    handle: OwnedSocketHandle,
+    deadline: Instant,
+    _keep_alive: Box<dyn Send>,
+}
+
 pub(crate) struct Reactor {
     notify: Arc<Notify>,
     iface: BufferInterface,
     socket_allocator: SocketAlloctor,
     tcp_listeners: SharedTcpListenerRegistry,
+    closing_tcp: mpsc::UnboundedSender<ClosingTcpSocket>,
 }
 
 async fn receive(
@@ -203,12 +213,14 @@ fn update_tcp_listeners(
     wakers
 }
 
+#[allow(clippy::too_many_arguments)]
 async fn run(
     mut async_iface: impl super::device::AsyncDevice,
     iface: BufferInterface,
     mut device: BufferDevice,
     socket_allocator: SocketAlloctor,
     tcp_listeners: SharedTcpListenerRegistry,
+    mut closing_tcp: mpsc::UnboundedReceiver<ClosingTcpSocket>,
     notify: Arc<Notify>,
     stopper: Arc<Notify>,
 ) -> io::Result<()> {
@@ -219,21 +231,36 @@ async fn run(
         .max_burst_size
         .unwrap_or(MAX_BURST_SIZE);
     let mut recv_buf = VecDeque::with_capacity(max_burst_size);
+    let mut closing = Vec::<ClosingTcpSocket>::new();
+    let mut closed = Vec::new();
     pin!(timer);
 
     loop {
         let packets = device.take_send_queue();
 
         async_iface.send_all(&mut iter(packets).map(Ok)).await?;
+        // Keep routing and stack leases until the last protocol output has
+        // left the device queue. Drop handles outside the socket-set lock.
+        closed.clear();
+        while let Ok(socket) = closing_tcp.try_recv() {
+            closing.push(socket);
+        }
 
         if recv_buf.is_empty() && device.need_wait() {
             let start = Instant::now();
-            let deadline = {
+            let mut deadline = {
                 iface
                     .lock()
                     .poll_delay(start, &socket_allocator.sockets().lock())
                     .unwrap_or(default_timeout)
             };
+            for socket in &closing {
+                deadline = deadline.min(if socket.deadline <= start {
+                    Duration::ZERO
+                } else {
+                    socket.deadline - start
+                });
+            }
 
             if deadline != Duration::ZERO {
                 timer
@@ -243,6 +270,7 @@ async fn run(
                     _ = &mut timer => {},
                     _ = receive(&mut async_iface,&mut recv_buf) => {}
                     _ = notify.notified() => {}
+                    Some(socket) = closing_tcp.recv() => closing.push(socket),
                     _ = stopper.notified() => break,
                 };
             }
@@ -290,6 +318,21 @@ async fn run(
         iface.poll(now, &mut device, &mut sockets);
         listener_wakers.extend(update_tcp_listeners(&mut listeners, &mut sockets));
 
+        if prepare_tcp_close(now, &closing, &mut sockets) {
+            iface.poll(now, &mut device, &mut sockets);
+        }
+        let mut index = 0;
+        while index < closing.len() {
+            let socket = sockets.get::<tcp::Socket>(*closing[index].handle);
+            // Closed can still have an RST to dispatch. TIME-WAIT must also
+            // remain available to acknowledge a retransmitted peer FIN.
+            if socket.state() == tcp::State::Closed && socket.remote_endpoint().is_none() {
+                closed.push(closing.swap_remove(index));
+            } else {
+                index += 1;
+            }
+        }
+
         // wake up all closed sockets (smoltcp seems have a bug that it doesn't wake up closed sockets)
         for (_, socket) in sockets.iter_mut() {
             if let Socket::Tcp(tcp) = socket
@@ -309,6 +352,30 @@ async fn run(
     Ok(())
 }
 
+fn prepare_tcp_close(
+    now: Instant,
+    closing: &[ClosingTcpSocket],
+    sockets: &mut SocketSet<'static>,
+) -> bool {
+    let mut changed = false;
+    for entry in closing {
+        let socket = sockets.get_mut::<tcp::Socket>(*entry.handle);
+        // There is no application reader anymore. Keep the receive window
+        // open so queued peer data cannot prevent its FIN from reaching us.
+        if socket.can_recv() {
+            let _ = socket.recv(|data| (data.len(), ()));
+            changed = true;
+        }
+        // The normal TCP timeout measures inactivity, so ACKs can extend it.
+        // A released application resource has an absolute teardown deadline.
+        if now >= entry.deadline && socket.state() != tcp::State::Closed {
+            socket.abort();
+            changed = true;
+        }
+    }
+    changed
+}
+
 impl Reactor {
     pub fn new(
         async_device: impl super::device::AsyncDevice,
@@ -321,12 +388,14 @@ impl Reactor {
         let notify = Arc::new(Notify::new());
         let socket_allocator = SocketAlloctor::new(buffer_size);
         let tcp_listeners = Arc::new(Mutex::new(TcpListenerRegistry::default()));
+        let (closing_tx, closing_rx) = mpsc::unbounded_channel();
         let fut = run(
             async_device,
             iface.clone(),
             device,
             socket_allocator.clone(),
             tcp_listeners.clone(),
+            closing_rx,
             notify.clone(),
             stopper,
         );
@@ -337,6 +406,7 @@ impl Reactor {
                 iface,
                 socket_allocator,
                 tcp_listeners,
+                closing_tcp: closing_tx,
             },
             fut,
         )
@@ -350,6 +420,21 @@ impl Reactor {
             |sockets: &mut smoltcp::iface::SocketSet<'_>| sockets.get_mut::<T>(handle),
         )
     }
+    pub(super) fn close_tcp(
+        &self,
+        handle: OwnedSocketHandle,
+        keep_alive: Box<dyn Send>,
+        deadline: Instant,
+    ) {
+        self.get_socket::<tcp::Socket>(*handle).close();
+        // The receiver and all closing sockets belong to run(), so stopping
+        // the reactor also releases them without a detached cleanup task.
+        let _ = self.closing_tcp.send(ClosingTcpSocket {
+            handle,
+            deadline,
+            _keep_alive: keep_alive,
+        });
+    }
     pub fn context(&self) -> MappedMutexGuard<'_, Context> {
         MutexGuard::map(self.iface.lock(), |iface| iface.context())
     }

easytier-core/src/gateway/smoltcp/tokio_smoltcp/socket.rs

@@ -97,6 +97,15 @@ pub struct TcpStream {
 }
 
 impl TcpStream {
+    /// Release the application stream while the reactor finishes TCP teardown.
+    pub(crate) fn close(self, keep_alive: impl Send + 'static) {
+        self.reactor.close_tcp(
+            self.handle,
+            Box::new(keep_alive),
+            smoltcp::time::Instant::now() + super::socket_allocator::TCP_TIMEOUT,
+        );
+    }
+
     pub(super) async fn connect(
         reactor: Arc<Reactor>,
         local_endpoint: IpEndpoint,
@@ -331,21 +340,21 @@ impl UdpSocket {
 
 #[cfg(test)]
 mod tests {
-    use std::time::Duration as StdDuration;
+    use std::{sync::Arc, time::Duration as StdDuration};
 
     use smoltcp::{
         iface::Config,
         phy::{DeviceCapabilities, Medium},
         socket::tcp,
         time::Duration,
-        wire::{HardwareAddress, Ipv4Address, TcpControl, TcpSeqNumber},
+        wire::{HardwareAddress, Ipv4Address, Ipv4Packet, TcpControl, TcpPacket, TcpSeqNumber},
     };
 
     use super::super::{
         Net, NetConfig, channel_device,
         test_utils::{TcpPackets, recv_tcp, recv_tcp_for_port},
     };
-    use super::{TCP_LISTENER_MAX_PENDING, TcpListener};
+    use super::{TCP_LISTENER_MAX_PENDING, TcpListener, TcpStream};
 
     const LISTEN_ADDR: Ipv4Address = Ipv4Address::new(10, 126, 126, 1);
     const LISTEN_PORT: u16 = 34569;
@@ -417,6 +426,136 @@ mod tests {
             .collect()
     }
 
+    async fn connected_stream(
+        net: &Net,
+        ingress: &tokio::sync::mpsc::Sender<std::io::Result<Vec<u8>>>,
+        egress: &mut tokio::sync::mpsc::Receiver<Vec<u8>>,
+    ) -> TcpStream {
+        let mut listener = net
+            .tcp_bind("10.126.126.1:34569".parse().unwrap())
+            .await
+            .unwrap();
+        let client_addr = Ipv4Address::new(10, 126, 126, 2);
+        let sequence = begin_handshake(ingress, egress, client_addr, 40000, 1000).await;
+        finish_handshake(ingress, client_addr, 40000, 1000, sequence).await;
+        tokio::time::timeout(StdDuration::from_secs(1), listener.accept())
+            .await
+            .unwrap()
+            .unwrap()
+            .0
+    }
+
+    #[tokio::test]
+    async fn closing_stream_retransmits_fin_and_retains_time_wait() {
+        let (net, ingress, mut egress) = test_net();
+        let stream = connected_stream(&net, &ingress, &mut egress).await;
+        let reactor = stream.reactor.clone();
+        let keep_alive = Arc::new(());
+        let retained = Arc::downgrade(&keep_alive);
+        stream.close(keep_alive);
+
+        let fin = recv_tcp(&mut egress).await;
+        assert_eq!(fin.control, TcpControl::Fin);
+        // Lose the first FIN. The application handle is gone, but the reactor
+        // must still retransmit the same sequence number.
+        let retransmit = tokio::time::timeout(StdDuration::from_secs(3), egress.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        let ipv4 = Ipv4Packet::new_checked(&retransmit).unwrap();
+        let tcp = TcpPacket::new_checked(ipv4.payload()).unwrap();
+        assert!(tcp.fin());
+        assert_eq!(tcp.seq_number(), fin.sequence);
+
+        let client_addr = Ipv4Address::new(10, 126, 126, 2);
+        let peer_fin = PACKETS.fin(client_addr, 40000, TcpSeqNumber(1001), fin.sequence + 1);
+        ingress.send(Ok(peer_fin.clone())).await.unwrap();
+        let ack = recv_tcp(&mut egress).await;
+        assert_eq!(ack.control, TcpControl::None);
+        assert_eq!(ack.ack_number, Some(TcpSeqNumber(1002)));
+        assert!(retained.upgrade().is_some());
+
+        // Lose our final ACK. TIME-WAIT must acknowledge the repeated FIN.
+        ingress.send(Ok(peer_fin)).await.unwrap();
+        let ack = recv_tcp(&mut egress).await;
+        assert_eq!(ack.control, TcpControl::None);
+        assert_eq!(ack.ack_number, Some(TcpSeqNumber(1002)));
+        tokio::time::timeout(StdDuration::from_secs(15), async {
+            while retained.upgrade().is_some() {
+                tokio::time::sleep(StdDuration::from_millis(10)).await;
+            }
+        })
+        .await
+        .expect("TIME-WAIT must eventually release the socket and its leases");
+        assert_eq!(
+            reactor.socket_allocator().sockets().lock().iter().count(),
+            0
+        );
+    }
+
+    #[tokio::test]
+    async fn closing_stream_deadline_sends_reset_and_releases_socket() {
+        let (net, ingress, mut egress) = test_net();
+        let stream = connected_stream(&net, &ingress, &mut egress).await;
+        let reactor = stream.reactor.clone();
+        let keep_alive = Arc::new(());
+        let retained = Arc::downgrade(&keep_alive);
+        // Exercise the same reactor deadline with a short test budget.
+        reactor.close_tcp(
+            stream.handle,
+            Box::new(keep_alive),
+            smoltcp::time::Instant::now() + Duration::from_millis(150),
+        );
+        let fin = recv_tcp(&mut egress).await;
+        assert_eq!(fin.control, TcpControl::Fin);
+        ingress
+            .send(Ok(PACKETS.ack(
+                Ipv4Address::new(10, 126, 126, 2),
+                40000,
+                TcpSeqNumber(1001),
+                fin.sequence + 1,
+            )))
+            .await
+            .unwrap();
+        // The peer ACKs our FIN but never closes its own half (FIN-WAIT-2).
+        let reset = recv_tcp(&mut egress).await;
+        assert_eq!(reset.control, TcpControl::Rst);
+        tokio::time::timeout(StdDuration::from_secs(1), async {
+            while retained.upgrade().is_some() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .unwrap();
+        assert_eq!(
+            reactor.socket_allocator().sockets().lock().iter().count(),
+            0
+        );
+    }
+
+    #[tokio::test]
+    async fn stopping_reactor_releases_closing_streams() {
+        let (net, ingress, mut egress) = test_net();
+        let stream = connected_stream(&net, &ingress, &mut egress).await;
+        let reactor = stream.reactor.clone();
+        let keep_alive = Arc::new(());
+        let retained = Arc::downgrade(&keep_alive);
+        stream.close(keep_alive);
+        assert_eq!(recv_tcp(&mut egress).await.control, TcpControl::Fin);
+        drop(net);
+        tokio::time::timeout(StdDuration::from_secs(1), async {
+            while retained.upgrade().is_some() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("stopping the reactor must release unfinished TCP teardown");
+        assert_eq!(
+            reactor.socket_allocator().sockets().lock().iter().count(),
+            0
+        );
+    }
+
     #[tokio::test]
     async fn duplicate_syn_does_not_consume_another_backlog_slot() {
         let (net, ingress, mut egress) = test_net();

easytier-core/src/gateway/smoltcp/tokio_smoltcp/test_utils.rs

@@ -114,6 +114,7 @@ pub(crate) struct TcpPacketSummary {
     pub(crate) control: TcpControl,
     pub(crate) sequence: TcpSeqNumber,
     pub(crate) dst_port: u16,
+    pub(crate) ack_number: Option<TcpSeqNumber>,
 }
 
 pub(crate) async fn recv_tcp(receiver: &mut mpsc::Receiver<Vec<u8>>) -> TcpPacketSummary {
@@ -133,6 +134,7 @@ pub(crate) async fn recv_tcp(receiver: &mut mpsc::Receiver<Vec<u8>>) -> TcpPacke
         control: repr.control,
         sequence: repr.seq_number,
         dst_port: repr.dst_port,
+        ack_number: repr.ack_number,
     }
 }
 

easytier-go/internal/artifact/provenance.go

@@ -3,6 +3,6 @@
 package artifact
 
 const (
-	Commit = "c51437c673a71f69b4b028c8d566a7b23b33e65c"
-	SHA256 = "17d6dd4561c1f1e818e7d70ed031c3cf8ea5dcfc1fd0461a6753f1f23f5fa6ab"
+	Commit = "5f1eaa7c9f9d3e89e9b75a6d87531a436fae40c4"
+	SHA256 = "de332a01eb4141e2e86ecee89500c40cc7d07a822a381868f95bb231bfc3739c"
 )

easytier-go/proto/api/config/api_config.pb.go

@@ -93,8 +93,11 @@ type InstanceConfigPatch struct {
 	VpnPortalClients       []*VpnPortalClientPatch      `protobuf:"bytes,15,rep,name=vpn_portal_clients,json=vpnPortalClients,proto3" json:"vpn_portal_clients,omitempty"`
 	ManagedCredentials     *manage.ManagedCredentialSet `protobuf:"bytes,16,opt,name=managed_credentials,json=managedCredentials,proto3" json:"managed_credentials,omitempty"`
 	PreferPeerRelay        *bool                        `protobuf:"varint,17,opt,name=prefer_peer_relay,json=preferPeerRelay,proto3,oneof" json:"prefer_peer_relay,omitempty"`
-	unknownFields          protoimpl.UnknownFields
-	sizeCache              protoimpl.SizeCache
+	// Applies with routes: true preserves an empty manual list; false restores
+	// automatic routes. Omission preserves legacy RoutePatch behavior.
+	EnableManualRoutes *bool `protobuf:"varint,18,opt,name=enable_manual_routes,json=enableManualRoutes,proto3,oneof" json:"enable_manual_routes,omitempty"`
+	unknownFields      protoimpl.UnknownFields
+	sizeCache          protoimpl.SizeCache
 }
 
 func (x *InstanceConfigPatch) Reset() {
@@ -246,6 +249,13 @@ func (x *InstanceConfigPatch) GetPreferPeerRelay() bool {
 	return false
 }
 
+func (x *InstanceConfigPatch) GetEnableManualRoutes() bool {
+	if x != nil && x.EnableManualRoutes != nil {
+		return *x.EnableManualRoutes
+	}
+	return false
+}
+
 type VpnPortalClientPatch struct {
 	state         protoimpl.MessageState        `protogen:"open.v1"`
 	Action        ConfigPatchAction             `protobuf:"varint,1,opt,name=action,proto3,enum=api.config.ConfigPatchAction" json:"action,omitempty"`
@@ -867,7 +877,7 @@ var File_api_config_proto protoreflect.FileDescriptor
 const file_api_config_proto_rawDesc = "" +
 	"\n" +
 	"\x10api_config.proto\x12\n" +
-	"api.config\x1a\fcommon.proto\x1a\tacl.proto\x1a\x12api_instance.proto\x1a\x10api_manage.proto\"\x86\t\n" +
+	"api.config\x1a\fcommon.proto\x1a\tacl.proto\x1a\x12api_instance.proto\x1a\x10api_manage.proto\"\xd6\t\n" +
 	"\x13InstanceConfigPatch\x12\x1f\n" +
 	"\bhostname\x18\x01 \x01(\tH\x00R\bhostname\x88\x01\x01\x12)\n" +
 	"\x04ipv4\x18\x02 \x01(\v2\x10.common.Ipv4InetH\x01R\x04ipv4\x88\x01\x01\x12)\n" +
@@ -889,7 +899,8 @@ const file_api_config_proto_rawDesc = "" +
 	"\x12disable_relay_data\x18\x0e \x01(\bH\aR\x10disableRelayData\x88\x01\x01\x12N\n" +
 	"\x12vpn_portal_clients\x18\x0f \x03(\v2 .api.config.VpnPortalClientPatchR\x10vpnPortalClients\x12Q\n" +
 	"\x13managed_credentials\x18\x10 \x01(\v2 .api.manage.ManagedCredentialSetR\x12managedCredentials\x12/\n" +
-	"\x11prefer_peer_relay\x18\x11 \x01(\bH\bR\x0fpreferPeerRelay\x88\x01\x01B\v\n" +
+	"\x11prefer_peer_relay\x18\x11 \x01(\bH\bR\x0fpreferPeerRelay\x88\x01\x01\x125\n" +
+	"\x14enable_manual_routes\x18\x12 \x01(\bH\tR\x12enableManualRoutes\x88\x01\x01B\v\n" +
 	"\t_hostnameB\a\n" +
 	"\x05_ipv4B\a\n" +
 	"\x05_ipv6B\x06\n" +
@@ -898,7 +909,8 @@ const file_api_config_proto_rawDesc = "" +
 	"\x16_ipv6_public_addr_autoB\x1a\n" +
 	"\x18_ipv6_public_addr_prefixB\x15\n" +
 	"\x13_disable_relay_dataB\x14\n" +
-	"\x12_prefer_peer_relay\"\x88\x01\n" +
+	"\x12_prefer_peer_relayB\x17\n" +
+	"\x15_enable_manual_routes\"\x88\x01\n" +
 	"\x14VpnPortalClientPatch\x125\n" +
 	"\x06action\x18\x01 \x01(\x0e2\x1d.api.config.ConfigPatchActionR\x06action\x129\n" +
 	"\x06client\x18\x02 \x01(\v2!.api.manage.VpnPortalClientConfigR\x06client\"x\n" +

easytier-go/proto/provenance.go

@@ -3,6 +3,6 @@
 package proto
 
 const (
-	EasyTierCommit = "c51437c673a71f69b4b028c8d566a7b23b33e65c"
-	SchemaSHA256   = "e49a4af09e5b66089eebb73fda78ffca7e8468a5a938cf7a593d689881686f0a"
+	EasyTierCommit = "5f1eaa7c9f9d3e89e9b75a6d87531a436fae40c4"
+	SchemaSHA256   = "33516bc93a3fd43df341ce081a904c1f8c5fd00ea5c33d78925e2359312a926c"
 )

easytier-go/proto/web/web.pb.go

@@ -95,8 +95,11 @@ type HeartbeatRequest struct {
 	SupportConfigSource     bool                   `protobuf:"varint,9,opt,name=support_config_source,json=supportConfigSource,proto3" json:"support_config_source,omitempty"`
 	FailedNetworkInstances  []*common.UUID         `protobuf:"bytes,10,rep,name=failed_network_instances,json=failedNetworkInstances,proto3" json:"failed_network_instances,omitempty"`
 	SupportHeartbeatPolicy  bool                   `protobuf:"varint,11,opt,name=support_heartbeat_policy,json=supportHeartbeatPolicy,proto3" json:"support_heartbeat_policy,omitempty"`
-	unknownFields           protoimpl.UnknownFields
-	sizeCache               protoimpl.SizeCache
+	// Supports explicit manual mode patches and typed empty-manual readback.
+	// Older peers silently ignore unknown patch fields; missing means false.
+	SupportManualRoutesPatch bool `protobuf:"varint,12,opt,name=support_manual_routes_patch,json=supportManualRoutesPatch,proto3" json:"support_manual_routes_patch,omitempty"`
+	unknownFields            protoimpl.UnknownFields
+	sizeCache                protoimpl.SizeCache
 }
 
 func (x *HeartbeatRequest) Reset() {
@@ -206,6 +209,13 @@ func (x *HeartbeatRequest) GetSupportHeartbeatPolicy() bool {
 	return false
 }
 
+func (x *HeartbeatRequest) GetSupportManualRoutesPatch() bool {
+	if x != nil {
+		return x.SupportManualRoutesPatch
+	}
+	return false
+}
+
 type HeartbeatResponse struct {
 	state               protoimpl.MessageState `protogen:"open.v1"`
 	HeartbeatIntervalMs *uint32                `protobuf:"varint,1,opt,name=heartbeat_interval_ms,json=heartbeatIntervalMs,proto3,oneof" json:"heartbeat_interval_ms,omitempty"`
@@ -346,7 +356,7 @@ const file_web_proto_rawDesc = "" +
 	"\fDeviceOsInfo\x12\x17\n" +
 	"\aos_type\x18\x01 \x01(\tR\x06osType\x12\x18\n" +
 	"\aversion\x18\x02 \x01(\tR\aversion\x12\"\n" +
-	"\fdistribution\x18\x03 \x01(\tR\fdistribution\"\x9d\x04\n" +
+	"\fdistribution\x18\x03 \x01(\tR\fdistribution\"\xdc\x04\n" +
 	"\x10HeartbeatRequest\x12+\n" +
 	"\n" +
 	"machine_id\x18\x01 \x01(\v2\f.common.UUIDR\tmachineId\x12%\n" +
@@ -362,7 +372,8 @@ const file_web_proto_rawDesc = "" +
 	"\x15support_config_source\x18\t \x01(\bR\x13supportConfigSource\x12F\n" +
 	"\x18failed_network_instances\x18\n" +
 	" \x03(\v2\f.common.UUIDR\x16failedNetworkInstances\x128\n" +
-	"\x18support_heartbeat_policy\x18\v \x01(\bR\x16supportHeartbeatPolicy\"\xb6\x01\n" +
+	"\x18support_heartbeat_policy\x18\v \x01(\bR\x16supportHeartbeatPolicy\x12=\n" +
+	"\x1bsupport_manual_routes_patch\x18\f \x01(\bR\x18supportManualRoutesPatch\"\xb6\x01\n" +
 	"\x11HeartbeatResponse\x127\n" +
 	"\x15heartbeat_interval_ms\x18\x01 \x01(\rH\x00R\x13heartbeatIntervalMs\x88\x01\x01\x125\n" +
 	"\x14heartbeat_timeout_ms\x18\x02 \x01(\rH\x01R\x12heartbeatTimeoutMs\x88\x01\x01B\x18\n" +

easytier-go/tests/host_dataplane_test.go

@@ -186,6 +186,143 @@ func testTCPDataPlane(
 	case <-ctx.Done():
 		t.Fatalf("blocked TCP read did not wake: %v", ctx.Err())
 	}
+
+	// FIN gate: Close must half-close the stream so the peer's read ends with
+	// io.EOF. Releasing the resource without shutdown_write drops the socket
+	// instead, which leaves the peer ESTABLISHED forever (the exit-node leak).
+	if err := serverConnection.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil {
+		t.Fatalf("set server read deadline: %v", err)
+	}
+	if _, err := serverConnection.Read(make([]byte, 1)); !errors.Is(err, io.EOF) {
+		t.Fatalf("server read after client close = %v, want io.EOF (FIN not sent)", err)
+	}
+	if err := serverConnection.Close(); err != nil {
+		t.Fatalf("server close after EOF: %v", err)
+	}
+}
+
+// Application deadlines must not prevent the guest from closing TCP.
+func TestTCPCloseAfterExpiredDeadlines(t *testing.T) {
+	past := time.Now().Add(-time.Second)
+	for name, expire := range map[string]func(net.Conn) error{
+		"write": func(c net.Conn) error { return c.SetWriteDeadline(past) },
+		"full":  func(c net.Conn) error { return c.SetDeadline(past) },
+	} {
+		t.Run(name, func(t *testing.T) {
+			ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
+			defer cancel()
+			host, server, client := startDataPlanePair(t, ctx)
+			defer host.Close(ctx)
+			defer server.Close(ctx)
+			defer client.Close(ctx)
+
+			listener := listenTCPEventually(t, ctx, server)
+			defer listener.Close()
+			accepted := make(chan connectionResult, 1)
+			go func() {
+				connection, err := listener.Accept()
+				accepted <- connectionResult{connection: connection, err: err}
+			}()
+			clientConnection := dialEventually(t, ctx, client,
+				fmt.Sprintf("10.144.0.101:%d", listener.Addr().(*net.TCPAddr).Port))
+			var serverConnection net.Conn
+			select {
+			case result := <-accepted:
+				if result.err != nil {
+					t.Fatalf("accept EasyTier TCP: %v", result.err)
+				}
+				serverConnection = result.connection
+			case <-ctx.Done():
+				t.Fatalf("accept EasyTier TCP: %v", ctx.Err())
+			}
+
+			if err := expire(clientConnection); err != nil {
+				t.Fatalf("set expired client deadline: %v", err)
+			}
+			if err := clientConnection.Close(); err != nil {
+				t.Fatalf("client close after expired deadline: %v", err)
+			}
+			if err := serverConnection.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil {
+				t.Fatalf("set server read deadline: %v", err)
+			}
+			if _, err := serverConnection.Read(make([]byte, 1)); !errors.Is(err, io.EOF) {
+				t.Fatalf("server read after expired-deadline close = %v, want io.EOF", err)
+			}
+			if err := expire(serverConnection); err != nil {
+				t.Fatalf("set expired server deadline: %v", err)
+			}
+			if err := serverConnection.Close(); err != nil {
+				t.Fatalf("server close after expired deadline: %v", err)
+			}
+		})
+	}
+}
+
+func TestTCPCloseWithBlockedWrite(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 45*time.Second)
+	defer cancel()
+	host, server, client := startDataPlanePair(t, ctx)
+	defer host.Close(ctx)
+	defer server.Close(ctx)
+	defer client.Close(ctx)
+
+	listener := listenTCPEventually(t, ctx, server)
+	defer listener.Close()
+	accepted := make(chan connectionResult, 1)
+	go func() {
+		connection, err := listener.Accept()
+		accepted <- connectionResult{connection: connection, err: err}
+	}()
+	clientConnection := dialEventually(t, ctx, client,
+		fmt.Sprintf("10.144.0.101:%d", listener.Addr().(*net.TCPAddr).Port))
+	defer clientConnection.Close()
+	var serverConnection net.Conn
+	select {
+	case result := <-accepted:
+		if result.err != nil {
+			t.Fatalf("accept EasyTier TCP: %v", result.err)
+		}
+		serverConnection = result.connection
+	case <-ctx.Done():
+		t.Fatalf("accept EasyTier TCP: %v", ctx.Err())
+	}
+	defer serverConnection.Close()
+	if err := serverConnection.SetReadDeadline(time.Now().Add(10 * time.Second)); err != nil {
+		t.Fatal(err)
+	}
+
+	written := make(chan error, 1)
+	go func() {
+		_, err := clientConnection.Write(make([]byte, 1024*1024))
+		written <- err
+	}()
+	// Observe real data before withholding reads to fill the TCP windows.
+	if _, err := io.ReadFull(serverConnection, make([]byte, 1)); err != nil {
+		t.Fatalf("read start of blocked write: %v", err)
+	}
+	select {
+	case err := <-written:
+		t.Fatalf("large write completed before Close: %v", err)
+	case <-time.After(100 * time.Millisecond):
+	}
+	if err := clientConnection.Close(); err != nil {
+		t.Fatalf("close with blocked write: %v", err)
+	}
+	select {
+	case err := <-written:
+		if !errors.Is(err, net.ErrClosed) {
+			t.Fatalf("blocked write after Close = %v, want net.ErrClosed", err)
+		}
+	case <-ctx.Done():
+		t.Fatalf("Close did not unblock Write: %v", ctx.Err())
+	}
+	// Bytes already accepted by TCP precede FIN; draining them must end in EOF.
+	if _, err := io.Copy(io.Discard, serverConnection); err != nil {
+		t.Fatalf("server read after Close = %v, want EOF", err)
+	}
+	if err := serverConnection.Close(); err != nil {
+		t.Fatalf("server close after EOF: %v", err)
+	}
 }
 
 func testUDPDataPlane(