fix(runtime): restore per-instance native thread configuration (#2658)

Restore per-instance Tokio executors for standard native instances,
honoring multi_thread and multi_thread_count with a two-worker minimum
in multi-thread mode. Preserve external execution for compact hosts.

Run construction, lifecycle operations, FFI data-plane tasks, and
hot-reloaded port forwards on the selected instance executor. Keep
shutdown cleanup running when its caller is cancelled, and release
owned executors without blocking their workers on runtime teardown.

Add runtime ownership, cancellation, cross-runtime connectivity, and
hot-reload regression coverage, and document the execution model.

Validation: all 45 PR checks passed, including linting, integration
tests, and cross-platform builds.

docs/core-architecture.md

@@ -349,6 +349,14 @@ A core instance:
 `CoreProcessRuntime`, plus optional protocol and platform capabilities. The
 bundle carries capabilities, not preconstructed portable managers.
 
+The standard native factory owns one continuously driven Tokio executor per
+instance. `multi_thread` selects a multi-thread or current-thread runtime;
+`multi_thread_count` sets the worker count with the historical minimum of two.
+Construction, start/stop, and FFI data-plane tasks use that executor, independent
+of the process management runtime. Dropping its Host owner shuts down the
+executor without blocking a worker on its own termination. Compact and WASI
+Hosts retain their externally supplied execution model.
+
 Each Module owns partial-start cleanup for its internal resources.
 `CoreInstance` has one outer cancellation and recovery path for the complete
 serial startup. `Running` therefore means the Host runtime and every enabled

easytier-core/src/gateway/port_forward.rs

@@ -100,6 +100,7 @@ where
 {
     operation: Mutex<()>,
     started: AtomicBool,
+    runtime_handle: Option<tokio::runtime::Handle>,
     runtime_config: CoreRuntimeConfigStore,
     data_plane: Arc<DataPlaneRuntime<H>>,
     host: Arc<H>,
@@ -124,10 +125,12 @@ where
         host: Arc<H>,
         socket_context: SocketContext,
         events: Arc<dyn CoreEventSink>,
+        runtime_handle: Option<tokio::runtime::Handle>,
     ) -> Arc<Self> {
         Arc::new(Self {
             operation: Mutex::new(()),
             started: AtomicBool::new(false),
+            runtime_handle,
             runtime_config,
             data_plane,
             host,
@@ -168,7 +171,22 @@ where
         Ok(())
     }
 
-    pub(crate) async fn reload(&self, cfgs: &[PortForwardConfig]) -> anyhow::Result<()> {
+    pub(crate) async fn reload(self: &Arc<Self>, cfgs: &[PortForwardConfig]) -> anyhow::Result<()> {
+        if let Some(runtime) = &self.runtime_handle {
+            let adapter = self.clone();
+            let cfgs = cfgs.to_vec();
+            // Binding and spawning must both happen on the instance executor,
+            // including reloads reached through management config patches.
+            return AbortOnDropHandle::new(
+                runtime.spawn(async move { adapter.reload_on_runtime(&cfgs).await }),
+            )
+            .await
+            .map_err(|error| anyhow::anyhow!("port-forward reload task failed: {error}"))?;
+        }
+        self.reload_on_runtime(cfgs).await
+    }
+
+    async fn reload_on_runtime(&self, cfgs: &[PortForwardConfig]) -> anyhow::Result<()> {
         let _operation = self.operation.lock().await;
         if !self.started.load(Ordering::Acquire) {
             return Ok(());

easytier-core/src/instance/lifecycle.rs

@@ -216,8 +216,25 @@ where
         self.packet_egress.stop().await;
     }
 
+    /// Returns the optional Host-owned executor for this instance.
+    pub fn runtime_handle(&self) -> Option<tokio::runtime::Handle> {
+        self.instance_runtime.runtime_handle()
+    }
+
     /// Starts the complete instance through one serial composition path.
     pub async fn start(self: &Arc<Self>) -> anyhow::Result<()> {
+        if let Some(runtime) = self.runtime_handle() {
+            let instance = self.clone();
+            return tokio_util::task::AbortOnDropHandle::new(
+                runtime.spawn(async move { instance.start_on_runtime().await }),
+            )
+            .await
+            .map_err(|error| anyhow::anyhow!("instance start task failed: {error}"))?;
+        }
+        self.start_on_runtime().await
+    }
+
+    async fn start_on_runtime(self: &Arc<Self>) -> anyhow::Result<()> {
         let _operation = self.operation.lock().await;
         let state = self.state();
         if state != CoreInstanceState::Created {
@@ -260,6 +277,21 @@ where
     }
 
     pub async fn stop(self: &Arc<Self>) {
+        self.cancel.cancel();
+        if let Some(runtime) = self.runtime_handle() {
+            let instance = self.clone();
+            // Cleanup must survive cancellation even before its first poll.
+            // Dropping a plain JoinHandle detaches rather than aborts the task.
+            runtime
+                .spawn(async move { instance.stop_on_runtime().await })
+                .await
+                .expect("instance stop task failed");
+        } else {
+            self.stop_on_runtime().await;
+        }
+    }
+
+    async fn stop_on_runtime(self: &Arc<Self>) {
         self.cancel.cancel();
         let mut recovery = self.recovery_guard();
         let _operation = self.operation.lock().await;

easytier-core/src/instance/manager.rs

@@ -479,9 +479,9 @@ where
     }
 
     pub fn data_plane_runtime_handle(&self, instance_id: &Uuid) -> Option<tokio::runtime::Handle> {
-        self.instance(*instance_id)?;
-        self.runtime_handle
-            .clone()
+        self.instance(*instance_id)?
+            .runtime_handle()
+            .or_else(|| self.runtime_handle.clone())
             .or_else(|| tokio::runtime::Handle::try_current().ok())
     }
 

easytier-core/src/instance/mod.rs

@@ -229,6 +229,12 @@ fn proxy_cidr_snapshot(config: &CoreInstanceRuntimeConfig) -> ProxyCidrSnapshot
 /// lifetime, such as a native packet interface.
 #[async_trait::async_trait]
 pub trait InstanceRuntimeHost: std::any::Any + Send + Sync + 'static {
+    /// Optional Host-owned executor used for construction, lifecycle, and data-plane tasks.
+    /// Hosts without a dedicated executor continue to use the caller's runtime.
+    fn runtime_handle(&self) -> Option<tokio::runtime::Handle> {
+        None
+    }
+
     async fn prepare(
         &self,
         packet_plane: Arc<CorePacketPlane>,
@@ -783,6 +789,7 @@ where
             host.clone(),
             direct_options.tcp_bind.context.clone(),
             events.clone(),
+            instance_runtime.runtime_handle(),
         );
         #[cfg(feature = "tcp-hole-punch")]
         let tcp_hole_punch = stun

easytier-core/src/instance/tests.rs

@@ -1513,6 +1513,162 @@ virtual_ip = "10.82.0.2/24"
         assert_eq!(runtime_host.shutdown_calls.load(Ordering::Relaxed), 1);
     }
 
+    #[test]
+    fn host_executor_routes_lifecycle_and_cancelled_start() {
+        struct ExecutorRuntimeHost {
+            handle: tokio::runtime::Handle,
+            block_prepare: bool,
+            prepare_started: Notify,
+            shutdown_calls: AtomicUsize,
+        }
+
+        #[async_trait]
+        impl InstanceRuntimeHost for ExecutorRuntimeHost {
+            fn runtime_handle(&self) -> Option<tokio::runtime::Handle> {
+                Some(self.handle.clone())
+            }
+
+            async fn prepare(
+                &self,
+                _packet_plane: Arc<CorePacketPlane>,
+            ) -> anyhow::Result<Option<Arc<dyn DhcpIpv4Host>>> {
+                assert_eq!(tokio::runtime::Handle::current().id(), self.handle.id());
+                self.prepare_started.notify_one();
+                if self.block_prepare {
+                    std::future::pending::<()>().await;
+                }
+                Ok(None)
+            }
+
+            async fn shutdown(&self) {
+                assert_eq!(tokio::runtime::Handle::current().id(), self.handle.id());
+                self.shutdown_calls.fetch_add(1, Ordering::Relaxed);
+            }
+        }
+
+        for block_prepare in [false, true] {
+            let executor = tokio::runtime::Builder::new_multi_thread()
+                .worker_threads(2)
+                .enable_all()
+                .build()
+                .unwrap();
+            let caller = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .unwrap();
+            let runtime_host = Arc::new(ExecutorRuntimeHost {
+                handle: executor.handle().clone(),
+                block_prepare,
+                prepare_started: Notify::new(),
+                shutdown_calls: AtomicUsize::new(0),
+            });
+            let instance = {
+                let _runtime = executor.enter();
+                let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16);
+                let mut adapters = adapters(None, Arc::new(packet_sink));
+                adapters.instance_runtime = runtime_host.clone();
+                CoreInstance::new(test_config("host-executor"), adapters).unwrap()
+            };
+            caller.block_on(async {
+                if block_prepare {
+                    let start = tokio::spawn({
+                        let instance = instance.clone();
+                        async move { instance.start().await }
+                    });
+                    tokio::time::timeout(
+                        Duration::from_secs(2),
+                        runtime_host.prepare_started.notified(),
+                    )
+                    .await
+                    .unwrap();
+                    start.abort();
+                    assert!(start.await.unwrap_err().is_cancelled());
+                } else {
+                    instance.start().await.unwrap();
+                    assert_eq!(instance.state(), CoreInstanceState::Running);
+                    instance.stop().await;
+                }
+                tokio::time::timeout(Duration::from_secs(2), async {
+                    while instance.state() != CoreInstanceState::Stopped {
+                        tokio::time::sleep(Duration::from_millis(1)).await;
+                    }
+                })
+                .await
+                .expect("lifecycle cleanup must complete on the Host executor");
+                assert_eq!(runtime_host.shutdown_calls.load(Ordering::Relaxed), 1);
+            });
+        }
+    }
+
+    #[test]
+    fn cancelled_stop_before_executor_poll_still_finishes_cleanup() {
+        struct ExecutorRuntimeHost {
+            handle: tokio::runtime::Handle,
+            shutdown_calls: AtomicUsize,
+        }
+
+        #[async_trait]
+        impl InstanceRuntimeHost for ExecutorRuntimeHost {
+            fn runtime_handle(&self) -> Option<tokio::runtime::Handle> {
+                Some(self.handle.clone())
+            }
+
+            async fn prepare(
+                &self,
+                _packet_plane: Arc<CorePacketPlane>,
+            ) -> anyhow::Result<Option<Arc<dyn DhcpIpv4Host>>> {
+                Ok(None)
+            }
+
+            async fn shutdown(&self) {
+                assert_eq!(tokio::runtime::Handle::current().id(), self.handle.id());
+                self.shutdown_calls.fetch_add(1, Ordering::Relaxed);
+            }
+        }
+
+        let executor = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        let runtime_host = Arc::new(ExecutorRuntimeHost {
+            handle: executor.handle().clone(),
+            shutdown_calls: AtomicUsize::new(0),
+        });
+        let instance = {
+            let _runtime = executor.enter();
+            let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16);
+            let mut adapters = adapters(None, Arc::new(packet_sink));
+            adapters.instance_runtime = runtime_host.clone();
+            CoreInstance::new(test_config("cancel-stop-before-poll"), adapters).unwrap()
+        };
+        executor.block_on(instance.start()).unwrap();
+        assert_eq!(instance.state(), CoreInstanceState::Running);
+
+        let caller = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        caller.block_on(async {
+            // The instance executor is not driven here, so the dispatched stop
+            // cannot be polled before its caller is cancelled.
+            let mut stop = Box::pin(instance.stop());
+            assert!(futures::poll!(&mut stop).is_pending());
+            drop(stop);
+        });
+        drop(caller);
+
+        executor.block_on(async {
+            tokio::time::timeout(Duration::from_secs(2), async {
+                while instance.state() != CoreInstanceState::Stopped {
+                    tokio::time::sleep(Duration::from_millis(1)).await;
+                }
+            })
+            .await
+            .expect("cancelled stop must still finish on the instance executor");
+        });
+        assert_eq!(runtime_host.shutdown_calls.load(Ordering::Relaxed), 1);
+    }
+
     #[cfg(feature = "management")]
     #[tokio::test]
     async fn cancelled_delete_finishes_stop_and_keeps_wait_blocked() {

easytier/Cargo.toml

@@ -70,6 +70,7 @@ tokio = { workspace = true, features = [
     "net",
     "process",
     "rt",
+    "rt-multi-thread",
     "signal",
     "sync",
     "time",

easytier/src/instance/composition.rs

@@ -31,7 +31,10 @@ use crate::{
     instance::runtime_host::NativeInstanceRuntimeHost,
 };
 
-use super::host::{NativeInstanceHost, native_instance_host};
+use super::{
+    host::{NativeInstanceHost, native_instance_host},
+    runtime_executor::NativeInstanceExecutor,
+};
 #[cfg(feature = "kcp")]
 use crate::gateway::kcp_proxy::KcpProxyService;
 #[cfg(feature = "quic")]
@@ -42,10 +45,11 @@ use easytier_core::gateway::proxy::wrapped_transport::WrappedTransportEngine;
 
 pub(crate) type NativeCoreInstance = CoreInstance<NativeInstanceHost>;
 
-pub(crate) fn compose_native_core_instance(
+pub(super) fn compose_native_core_instance(
     toml_config: TomlConfig,
     process_runtime: Arc<CoreProcessRuntime>,
     compact_runtime: bool,
+    executor: Option<NativeInstanceExecutor>,
 ) -> anyhow::Result<Arc<NativeCoreInstance>> {
     let host_config = if compact_runtime {
         compact_runtime_core_host_config()
@@ -58,7 +62,7 @@ pub(crate) fn compose_native_core_instance(
         &normalized,
         &host_config,
     ));
-    let runtime_host = NativeInstanceRuntimeHost::new(global_ctx.clone());
+    let runtime_host = NativeInstanceRuntimeHost::new_with_executor(global_ctx.clone(), executor);
     let mut adapters = runtime_core_host_adapters_with_packet_egress_and_config(
         global_ctx.clone(),
         process_runtime,

easytier/src/instance/factory.rs

@@ -5,7 +5,7 @@ use easytier_core::instance::manager::InstanceManager;
 #[cfg(feature = "management-rpc")]
 use easytier_core::management::ProcessRuntimeProvider;
 use easytier_core::{
-    config::toml::TomlConfig,
+    config::toml::{ConfigLoader as _, TomlConfig},
     instance::{CoreInstance, manager::InstanceFactory},
     process_runtime::CoreProcessRuntime,
 };
@@ -14,7 +14,7 @@ use crate::common::global_ctx::EventBusSubscriber;
 
 use super::{
     composition::compose_native_core_instance, host::NativeInstanceHost,
-    runtime_host::NativeInstanceRuntimeHost,
+    runtime_executor::NativeInstanceExecutor, runtime_host::NativeInstanceRuntimeHost,
 };
 
 pub type NativeCoreInstance = CoreInstance<NativeInstanceHost>;
@@ -142,14 +142,25 @@ impl InstanceFactory for NativeInstanceFactory {
         config: TomlConfig,
         (): Self::CreateContext,
     ) -> Result<Arc<Self::Instance>, Self::Error> {
-        let _runtime = self
-            .runtime_handle
+        let executor = if self.compact_runtime {
+            None
+        } else {
+            let flags = config.get_flags();
+            Some(NativeInstanceExecutor::new(
+                flags.multi_thread,
+                flags.multi_thread_count,
+            )?)
+        };
+        let runtime_handle = executor
             .as_ref()
-            .map(tokio::runtime::Handle::enter);
+            .map(NativeInstanceExecutor::handle)
+            .or_else(|| self.runtime_handle.clone());
+        let _runtime = runtime_handle.as_ref().map(tokio::runtime::Handle::enter);
         let instance = compose_native_core_instance(
             config,
             self.process_runtime.clone(),
             self.compact_runtime,
+            executor,
         )?;
         #[cfg(feature = "logging")]
         if self.log_cli_events {
@@ -170,10 +181,297 @@ impl ProcessRuntimeProvider for NativeInstanceFactory {
 
 #[cfg(test)]
 mod tests {
-    use easytier_core::{config::toml::ConfigLoader as _, instance::CoreInstanceState};
+    use std::{future::pending, time::Duration};
+
+    use easytier_core::instance::{CoreInstanceState, manager::ConfigFileControl};
+    use tokio::runtime::RuntimeFlavor;
 
     use super::*;
 
+    fn isolated_config(multi_thread: bool, count: u32) -> TomlConfig {
+        let config = TomlConfig::default();
+        let mut flags = config.get_flags();
+        flags.no_tun = true;
+        flags.enable_ipv6 = false;
+        flags.disable_upnp = true;
+        flags.disable_p2p = true;
+        flags.disable_tcp_hole_punching = true;
+        flags.disable_udp_hole_punching = true;
+        flags.multi_thread = multi_thread;
+        flags.multi_thread_count = count;
+        config.set_flags(flags);
+        config.set_listeners(Vec::new());
+        config.set_stun_servers(Some(Vec::new()));
+        config.set_stun_servers_v6(Some(Vec::new()));
+        config.set_tcp_stun_servers(Some(Vec::new()));
+        config
+    }
+
+    #[tokio::test]
+    async fn manager_honors_per_instance_runtime_configuration_and_teardown() {
+        let caller = tokio::runtime::Handle::current();
+        let manager = native_instance_manager_with_runtime(caller.clone());
+        let mut runtimes = Vec::new();
+        let mut ids = Vec::new();
+        for (multi_thread, count, workers) in [(false, 8, 1), (true, 4, 4)] {
+            let config = isolated_config(multi_thread, count);
+            let id = manager
+                .run_network_instance(config, ConfigFileControl::STATIC_CONFIG)
+                .unwrap();
+            let runtime = manager.data_plane_runtime_handle(&id).unwrap();
+            assert_ne!(runtime.id(), caller.id());
+            assert_eq!(runtime.metrics().num_workers(), workers);
+            assert_eq!(
+                runtime.runtime_flavor(),
+                if multi_thread {
+                    RuntimeFlavor::MultiThread
+                } else {
+                    RuntimeFlavor::CurrentThread
+                }
+            );
+            ids.push(id);
+            runtimes.push(runtime);
+        }
+        assert_ne!(runtimes[0].id(), runtimes[1].id());
+        tokio::time::timeout(Duration::from_secs(5), async {
+            while manager
+                .instances()
+                .iter()
+                .any(|instance| !instance.is_ready())
+            {
+                tokio::time::sleep(Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .expect("instances must start on their own executors");
+
+        // Shared process resources must continue to work across distinct executors.
+        let single = manager.instance(ids[0]).unwrap();
+        let multi = manager.instance(ids[1]).unwrap();
+        multi
+            .add_connector(format!("ring://{}", ids[0]).parse().unwrap())
+            .unwrap();
+        tokio::time::timeout(Duration::from_secs(10), async {
+            loop {
+                if single.connected_peers().await.contains(&multi.peer_id())
+                    && multi.connected_peers().await.contains(&single.peer_id())
+                {
+                    break;
+                }
+                tokio::time::sleep(Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .expect("single- and multi-thread instances must connect through the shared ring registry");
+        drop(single);
+        drop(multi);
+
+        let instances = manager.instances();
+        let tasks: Vec<_> = runtimes
+            .iter()
+            .map(|runtime| runtime.spawn(pending::<()>()))
+            .collect();
+        manager.delete_network_instances(ids.clone()).await.unwrap();
+        assert!(
+            instances
+                .iter()
+                .all(|instance| instance.state() == CoreInstanceState::Stopped)
+        );
+        assert!(
+            ids.iter()
+                .all(|id| manager.data_plane_runtime_handle(id).is_none())
+        );
+        drop(instances);
+        for task in tasks {
+            assert!(
+                tokio::time::timeout(Duration::from_secs(5), task)
+                    .await
+                    .expect("deleting the last instance owner must shut down its executor")
+                    .unwrap_err()
+                    .is_cancelled()
+            );
+        }
+    }
+
+    #[test]
+    fn direct_instance_outlives_its_callers_runtime() {
+        for multi_thread in [false, true] {
+            let caller = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .unwrap();
+            // Standard native construction no longer needs an entered caller runtime.
+            let instance = create_native_instance(isolated_config(multi_thread, 3)).unwrap();
+            caller.block_on(instance.start()).unwrap();
+            drop(caller);
+            let executor = instance.runtime_handle().unwrap();
+            let caller = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .unwrap();
+            caller.block_on(async {
+                tokio::time::timeout(
+                    Duration::from_secs(5),
+                    executor.spawn(async {
+                        tokio::time::sleep(Duration::from_millis(10)).await;
+                    }),
+                )
+                .await
+                .unwrap()
+                .unwrap();
+                instance.stop().await;
+            });
+            assert_eq!(instance.state(), CoreInstanceState::Stopped);
+            drop(instance);
+        }
+    }
+
+    #[cfg(all(feature = "smoltcp", feature = "management"))]
+    #[test]
+    fn hot_added_port_forwards_outlive_the_callers_runtime() {
+        use easytier_core::{
+            config::{gateway::PortForwardConfig, runtime::CoreInstanceRuntimeConfig},
+            instance::CoreInstanceConfig,
+        };
+        use easytier_proto::api::config::{
+            ConfigPatchAction, InstanceConfigPatch, PortForwardPatch,
+        };
+        use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
+
+        for multi_thread in [false, true] {
+            for via_patch in [false, true] {
+                let observer = tokio::runtime::Builder::new_current_thread()
+                    .enable_all()
+                    .build()
+                    .unwrap();
+                let (tcp_echo, udp_echo) = observer.block_on(async {
+                    (
+                        tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(),
+                        tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap(),
+                    )
+                });
+                let tcp_reservation = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+                let udp_reservation = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
+                let forwards = vec![
+                    PortForwardConfig {
+                        bind_addr: tcp_reservation.local_addr().unwrap(),
+                        dst_addr: tcp_echo.local_addr().unwrap(),
+                        proto: "tcp".to_owned(),
+                    },
+                    PortForwardConfig {
+                        bind_addr: udp_reservation.local_addr().unwrap(),
+                        dst_addr: udp_echo.local_addr().unwrap(),
+                        proto: "udp".to_owned(),
+                    },
+                ];
+                let config = isolated_config(multi_thread, 2);
+                // Local virtual-IP forwarding uses native loopback endpoints,
+                // avoiding TUN devices, network namespaces and remote peers.
+                config.set_ipv4(Some("127.0.0.1/32".parse().unwrap()));
+                let instance = create_native_instance(config.clone()).unwrap();
+                let caller = tokio::runtime::Builder::new_current_thread()
+                    .enable_all()
+                    .build()
+                    .unwrap();
+                caller.block_on(async {
+                    instance.start().await.unwrap();
+                    drop(tcp_reservation);
+                    drop(udp_reservation);
+                    if via_patch {
+                        easytier_core::management::apply_config_patch(
+                            &instance,
+                            InstanceConfigPatch {
+                                port_forwards: forwards
+                                    .iter()
+                                    .cloned()
+                                    .map(|forward| PortForwardPatch {
+                                        action: ConfigPatchAction::Add as i32,
+                                        cfg: Some(forward.into()),
+                                    })
+                                    .collect(),
+                                ..Default::default()
+                            },
+                            None,
+                        )
+                        .await
+                        .unwrap();
+                    } else {
+                        let mut normalized = CoreInstanceConfig::from_toml(&config).unwrap();
+                        normalized.connectivity.runtime.gateway.port_forwards = forwards.clone();
+                        instance
+                            .update_runtime_config(CoreInstanceRuntimeConfig {
+                                services: normalized.connectivity.runtime,
+                                peer: Arc::new(normalized.peer.snapshot),
+                            })
+                            .await
+                            .unwrap();
+                    }
+                });
+                // No accept/receive task or reactor may depend on this runtime.
+                drop(caller);
+                assert_eq!(instance.state(), CoreInstanceState::Running);
+
+                observer.block_on(async {
+                    tokio::time::timeout(Duration::from_secs(5), async {
+                        let tcp_response = tokio::spawn(async move {
+                            let (mut stream, _) = tcp_echo.accept().await.unwrap();
+                            let mut request = [0; 4];
+                            stream.read_exact(&mut request).await.unwrap();
+                            stream.write_all(&request).await.unwrap();
+                        });
+                        let udp_response = tokio::spawn(async move {
+                            let mut request = [0; 4];
+                            let (len, source) = udp_echo.recv_from(&mut request).await.unwrap();
+                            udp_echo.send_to(&request[..len], source).await.unwrap();
+                        });
+                        let mut stream = tokio::net::TcpStream::connect(forwards[0].bind_addr)
+                            .await
+                            .unwrap();
+                        stream.write_all(b"ping").await.unwrap();
+                        let mut reply = [0; 4];
+                        stream.read_exact(&mut reply).await.unwrap();
+                        assert_eq!(&reply, b"ping");
+                        tcp_response.await.unwrap();
+
+                        let socket = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
+                        socket
+                            .send_to(b"pong", forwards[1].bind_addr)
+                            .await
+                            .unwrap();
+                        let (len, _) = socket.recv_from(&mut reply).await.unwrap();
+                        assert_eq!(&reply[..len], b"pong");
+                        udp_response.await.unwrap();
+                    })
+                    .await
+                    .expect("hot-added TCP/UDP forwarding must survive caller-runtime shutdown");
+                    instance.stop().await;
+                });
+                assert_eq!(instance.state(), CoreInstanceState::Stopped);
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn compact_manager_retains_the_supplied_runtime() {
+        let caller = tokio::runtime::Handle::current();
+        let manager = native_compact_instance_manager_with_runtime(caller.clone());
+        let instance = manager.create(isolated_config(true, 8), ()).unwrap();
+        assert!(instance.runtime_handle().is_none());
+        assert_eq!(
+            manager
+                .data_plane_runtime_handle(&instance.instance_id())
+                .unwrap()
+                .id(),
+            caller.id()
+        );
+        instance.start().await.unwrap();
+        manager
+            .delete_network_instances([instance.instance_id()])
+            .await
+            .unwrap();
+        assert_eq!(instance.state(), CoreInstanceState::Stopped);
+    }
+
     #[tokio::test]
     async fn core_manager_stores_and_runs_native_core_instance_directly() {
         let factory = NativeInstanceFactory::new(CoreProcessRuntime::new());

easytier/src/instance/mod.rs

@@ -7,6 +7,7 @@ pub(crate) mod config_storage;
 pub mod dns_server;
 pub mod factory;
 pub mod host;
+mod runtime_executor;
 pub(crate) mod runtime_host;
 #[cfg(test)]
 pub(crate) mod test_instance;

easytier/src/instance/runtime_executor.rs

@@ -0,0 +1,109 @@
+//! Native ownership and continuous driving of one instance's Tokio runtime.
+
+use std::sync::mpsc;
+
+use tokio::{
+    runtime::{Builder, Handle},
+    sync::oneshot,
+};
+
+pub(super) struct NativeInstanceExecutor {
+    handle: Handle,
+    // Dropping the sender stops the driver, including on construction failure.
+    _shutdown: oneshot::Sender<()>,
+}
+
+impl NativeInstanceExecutor {
+    pub(super) fn new(multi_thread: bool, thread_count: u32) -> anyhow::Result<Self> {
+        let (ready_tx, ready_rx) = mpsc::sync_channel(1);
+        let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
+        std::thread::Builder::new()
+            .name("easytier-instance".to_owned())
+            .spawn(move || {
+                let mut builder = if multi_thread {
+                    let mut builder = Builder::new_multi_thread();
+                    builder.worker_threads(2.max(thread_count as usize));
+                    builder
+                } else {
+                    Builder::new_current_thread()
+                };
+                let runtime = match builder.enable_all().build() {
+                    Ok(runtime) => runtime,
+                    Err(error) => {
+                        let _ = ready_tx.send(Err(error));
+                        return;
+                    }
+                };
+                if ready_tx.send(Ok(runtime.handle().clone())).is_ok() {
+                    runtime.block_on(async {
+                        let _ = shutdown_rx.await;
+                    });
+                }
+                // Never block an instance worker waiting for its own shutdown.
+                // Blocking Host operations may finish after the owner is dropped.
+                runtime.shutdown_background();
+            })?;
+        let handle = ready_rx.recv()??;
+        Ok(Self {
+            handle,
+            _shutdown: shutdown_tx,
+        })
+    }
+
+    pub(super) fn handle(&self) -> Handle {
+        self.handle.clone()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::{future::pending, time::Duration};
+
+    use tokio::runtime::RuntimeFlavor;
+
+    use super::*;
+
+    #[tokio::test]
+    async fn honors_runtime_flavor_and_worker_count() {
+        for (multi_thread, count, expected_workers) in [
+            (false, 8, 1),
+            (true, 0, 2),
+            (true, 1, 2),
+            (true, 2, 2),
+            (true, 4, 4),
+        ] {
+            let executor = NativeInstanceExecutor::new(multi_thread, count).unwrap();
+            let handle = executor.handle();
+            assert_eq!(handle.metrics().num_workers(), expected_workers);
+            assert_eq!(
+                handle.runtime_flavor(),
+                if multi_thread {
+                    RuntimeFlavor::MultiThread
+                } else {
+                    RuntimeFlavor::CurrentThread
+                }
+            );
+            let caller_thread = std::thread::current().id();
+            let task_thread = handle
+                .spawn(async { std::thread::current().id() })
+                .await
+                .unwrap();
+            assert_ne!(task_thread, caller_thread);
+        }
+    }
+
+    #[tokio::test]
+    async fn dropping_owner_from_a_worker_shuts_down_tasks() {
+        for multi_thread in [false, true] {
+            let executor = NativeInstanceExecutor::new(multi_thread, 2).unwrap();
+            let handle = executor.handle();
+            let task = handle.spawn(pending::<()>());
+            handle.spawn(async move { drop(executor) });
+            let error = tokio::time::timeout(Duration::from_secs(5), task)
+                .await
+                .expect("runtime must shut down without joining its own worker")
+                .unwrap_err();
+            assert!(error.is_cancelled());
+        }
+    }
+}

easytier/src/instance/runtime_host.rs

@@ -25,6 +25,7 @@ mod tun_runtime;
 #[path = "runtime_host/tun_mobile.rs"]
 mod tun_runtime;
 
+use super::runtime_executor::NativeInstanceExecutor;
 use event_journal::EventJournal;
 #[cfg(feature = "tun")]
 use magic_dns::MagicDnsRuntime;
@@ -36,10 +37,19 @@ pub(crate) struct NativeInstanceRuntimeHost {
     cancel: CancellationToken,
     event_journal: EventJournal,
     tun: NativeTunRuntime,
+    executor: Option<NativeInstanceExecutor>,
 }
 
 impl NativeInstanceRuntimeHost {
+    #[cfg(test)]
     pub(crate) fn new(global_ctx: ArcGlobalCtx) -> Arc<Self> {
+        Self::new_with_executor(global_ctx, None)
+    }
+
+    pub(super) fn new_with_executor(
+        global_ctx: ArcGlobalCtx,
+        executor: Option<NativeInstanceExecutor>,
+    ) -> Arc<Self> {
         let cancel = CancellationToken::new();
         let tun = NativeTunRuntime::new(global_ctx.clone(), cancel.clone());
         let event_journal = EventJournal::new(&global_ctx);
@@ -49,6 +59,7 @@ impl NativeInstanceRuntimeHost {
             operation: Arc::new(Mutex::new(())),
             cancel,
             tun,
+            executor,
         })
     }
 

easytier/src/instance/runtime_host/implementation.rs

@@ -10,6 +10,10 @@ use super::NativeInstanceRuntimeHost;
 
 #[async_trait::async_trait]
 impl InstanceRuntimeHost for NativeInstanceRuntimeHost {
+    fn runtime_handle(&self) -> Option<tokio::runtime::Handle> {
+        self.executor.as_ref().map(|executor| executor.handle())
+    }
+
     async fn prepare(
         &self,
         packet_plane: Arc<CorePacketPlane>,