EasyTier/easytier-core/src/instance/tests.rs

use std::sync::Arc;

use async_trait::async_trait;

use super::*;
use crate::{
    config::toml::ConfigLoader as _,
    listener::transport::TransportListenerConfig,
    socket::{
        SocketContext, SocketListener,
        udp::{UdpSessionAcceptKind, UdpSessionProtocol},
    },
};

struct TestServerProtocol;

#[async_trait]
impl ServerProtocolUpgrader<()> for TestServerProtocol {
    fn supports_scheme(&self, scheme: &str) -> bool {
        matches!(scheme, "ws" | "wss" | "wg" | "quic" | "faketcp" | "unix")
    }

    async fn upgrade_tcp(
        &self,
        _socket: (),
        _local_url: Url,
    ) -> anyhow::Result<crate::connectivity::protocol::ServerProtocolUpgrade> {
        unreachable!()
    }

    async fn upgrade_udp(
        &self,
        _session: crate::socket::udp::UdpSession,
        _local_url: Url,
        _admission: Option<crate::connectivity::protocol::ServerProtocolAdmission>,
    ) -> anyhow::Result<crate::connectivity::protocol::ServerProtocolUpgrade> {
        unreachable!()
    }

    async fn upgrade_byte_stream(
        &self,
        _socket: (),
        _local_url: Url,
        _remote_url: Option<Url>,
    ) -> anyhow::Result<crate::connectivity::protocol::ServerProtocolUpgrade> {
        unreachable!()
    }
}

struct TestExternalListenerFactory;

impl ExternalListenerFactory<()> for TestExternalListenerFactory {
    fn supports_scheme(&self, scheme: &str) -> bool {
        matches!(scheme, "faketcp" | "unix")
    }

    fn create(&self, _request: ExternalListenerRequest) -> Box<dyn SocketListener<Accepted = ()>> {
        unreachable!()
    }
}

#[test]
fn runtime_updates_retain_core_owned_peer_identity() {
    let mut snapshot = Arc::new(PeerRuntimeSnapshot::default());
    Arc::make_mut(&mut snapshot).runtime.core.node.peer_id = Some(17);
    Arc::make_mut(&mut snapshot).runtime.core.node.instance_id = Some([1; 16]);
    let submitted = snapshot.clone();

    retain_core_peer_identity(&mut snapshot, 23, Some([2; 16]));

    assert_eq!(snapshot.runtime.core.node.peer_id, Some(23));
    assert_eq!(snapshot.runtime.core.node.instance_id, Some([2; 16]));
    assert_eq!(submitted.runtime.core.node.peer_id, Some(17));
    assert_eq!(submitted.runtime.core.node.instance_id, Some([1; 16]));
}

#[test]
fn core_plans_transport_and_external_listener_capabilities() {
    let self_id = uuid::Uuid::new_v4();
    let config = ListenerRuntimeConfig::new(
        [
            "tcp://127.0.0.1:1",
            "udp://127.0.0.1:2",
            "ws://127.0.0.1:3",
            "wg://127.0.0.1:4",
            "quic://127.0.0.1:5",
            "faketcp://127.0.0.1:6",
            "unix:///tmp/easytier-test",
            "http://127.0.0.1:7",
        ]
        .into_iter()
        .map(str::parse)
        .collect::<Result<Vec<_>, _>>()
        .unwrap(),
        false,
        SocketContext::default().with_socket_mark(Some(7)),
    );

    let plan = prepare_listener_plan::<(), ()>(
        Some(&config),
        self_id,
        Some(&TestServerProtocol),
        Some(&TestExternalListenerFactory),
    )
    .unwrap();

    assert_eq!(plan.transports.len(), 6);
    assert_eq!(plan.external.len(), 2);
    assert_eq!(plan.failures.len(), 1);
    assert_eq!(
        plan.transports[0].url(),
        &crate::listener::plan::ring_listener_url(self_id)
    );
    assert!(matches!(
        &plan.transports[4],
        TransportListenerConfig::Udp {
            accept_kind: UdpSessionAcceptKind::Classified(UdpSessionProtocol::WireGuard),
            ..
        }
    ));
    assert!(matches!(
        &plan.transports[5],
        TransportListenerConfig::Udp {
            accept_kind: UdpSessionAcceptKind::Classified(UdpSessionProtocol::Quic),
            ..
        }
    ));
    assert_eq!(plan.external[0].0.url.scheme(), "faketcp");
    assert_eq!(plan.external[0].1.socket_mark, Some(7));
}

#[test]
fn unsupported_protocol_listener_becomes_a_plan_failure() {
    let config = ListenerRuntimeConfig::new(
        vec!["wg://127.0.0.1:11011".parse().unwrap()],
        false,
        SocketContext::default(),
    );

    let plan =
        prepare_listener_plan::<(), ()>(Some(&config), uuid::Uuid::new_v4(), None, None).unwrap();

    assert_eq!(plan.transports.len(), 1);
    assert!(plan.external.is_empty());
    assert_eq!(plan.failures.len(), 1);
}

#[test]
fn raw_unix_listener_does_not_require_a_server_protocol() {
    let config = ListenerRuntimeConfig::new(
        vec!["unix:///tmp/easytier-test".parse().unwrap()],
        false,
        SocketContext::default(),
    );

    let plan = prepare_listener_plan::<(), ()>(
        Some(&config),
        uuid::Uuid::new_v4(),
        None,
        Some(&TestExternalListenerFactory),
    )
    .unwrap();

    assert_eq!(plan.transports.len(), 1);
    assert_eq!(plan.external.len(), 1);
    assert!(plan.failures.is_empty());
    assert_eq!(plan.external[0].0.url.scheme(), "unix");
}

#[test]
fn core_instance_config_round_trips_as_normalized_json() {
    let mut core = crate::config::CoreConfig::default();
    core.peer_policy.encryption_required = false;
    core.peer_policy.p2p_enabled = false;
    let peer = crate::peers::peer_manager::PortablePeerManagerConfig::new(
        crate::config::peers::PeerRuntimeConfig {
            core,
            network_identity: crate::config::NetworkIdentity {
                network_name: "default".to_owned(),
                network_secret: Some("test".to_owned()),
                network_secret_digest: None,
            },
            stun_info: crate::proto::common::StunInfo::default(),
            feature_flags: crate::proto::common::PeerFeatureFlag::default(),
            secure_mode: None,
            host_routing: crate::config::peers::HostRoutingPolicy::default(),
        },
    );
    let config = CoreInstanceConfig {
        instance_name: String::new(),
        peer,
        connectivity: CoreConnectivityConfig::default(),
    };

    let mut config = config;
    config.connectivity.direct.testing = true;
    let encoded = serde_json::to_value(&config).unwrap();
    assert!(encoded["connectivity"]["direct"].get("testing").is_none());
    let decoded: CoreInstanceConfig = serde_json::from_value(encoded.clone()).unwrap();

    assert!(!decoded.connectivity.direct.testing);
    assert!(decoded.connectivity.startup_plan.gateway);
    assert_eq!(serde_json::to_value(&decoded).unwrap(), encoded);

    let mut legacy = encoded;
    legacy.as_object_mut().unwrap().remove("instance_name");
    let decoded: CoreInstanceConfig = serde_json::from_value(legacy).unwrap();
    assert_eq!(decoded.instance_name, "default");
}

#[test]
fn wasi_create_config_uses_shared_toml() {
    let fixture = include_bytes!("../../testdata/wasi_core_instance_create.json");
    let mut create =
        serde_json::from_slice::<crate::wasi::schema::WasiCoreInstanceCreateConfig>(fixture)
            .unwrap();
    create.validate().unwrap();
    let config = create.parse_config().unwrap();
    let normalized = CoreInstanceConfig::from_toml(&config).unwrap();

    assert_eq!(
        normalized.peer.snapshot.runtime.core.node.instance_id,
        Some(*config.get_id().as_bytes())
    );
    assert!(normalized.peer.snapshot.flags.disable_p2p);
    create.version += 1;
    assert!(create.validate().is_err());
}

#[test]
fn core_instance_config_validation_rejects_invalid_acl_whitelist() {
    let peer = crate::peers::peer_manager::PortablePeerManagerConfig::new(
        crate::config::peers::PeerRuntimeConfig {
            core: crate::config::CoreConfig::default(),
            network_identity: crate::config::NetworkIdentity {
                network_name: "default".to_owned(),
                network_secret: Some("test".to_owned()),
                network_secret_digest: None,
            },
            stun_info: crate::proto::common::StunInfo::default(),
            feature_flags: crate::proto::common::PeerFeatureFlag::default(),
            secure_mode: None,
            host_routing: crate::config::peers::HostRoutingPolicy::default(),
        },
    );
    let mut config = CoreInstanceConfig {
        instance_name: String::new(),
        peer,
        connectivity: CoreConnectivityConfig::default(),
    };
    config.connectivity.runtime.acl.tcp_whitelist = vec!["9000-8000".to_owned()];

    let error = validate_core_instance_config(&config).unwrap_err();

    assert!(error.to_string().contains("Start port must be <= end port"));
}

mod portable_runtime {
    use std::{
        net::{IpAddr, Ipv6Addr, SocketAddr},
        sync::{
            Arc,
            atomic::{AtomicBool, AtomicUsize, Ordering},
        },
        time::Duration,
    };

    use tokio::sync::Notify;
    use tokio_util::task::AbortOnDropHandle;

    #[cfg(feature = "proxy-packet")]
    use std::sync::Mutex as StdMutex;

    use super::*;
    use crate::{
        config::peers::{HostRoutingPolicy, PeerRuntimeConfig},
        config::runtime::CoreInstanceRuntimeConfig,
        config::{
            CoreConfig, IpPrefix, NetworkIdentity, ProxyNetworkConfig, gateway::PortForwardConfig,
        },
        connectivity::manual::{ManualConnectorHost, ManualInterfaceAddrs},
        gateway::proxy::wrapped_transport::{
            WrappedTransportEngine, WrappedTransportEngineStart, WrappedTransportEngines,
            WrappedTransportRole,
        },
        host::testkit::{TestDns, TestHost, TestTcpSocket},
        listener::transport::AcceptedTransport,
        peers::peer_manager::PortablePeerManagerConfig,
        proto::{common::StunInfo, peer_rpc::GetIpListResponse},
        socket::{SocketContext, udp::PreferredIpv6Source},
    };

    #[cfg(feature = "proxy-packet")]
    use crate::gateway::proxy::wrapped_transport::WrappedTransportKind;

    #[async_trait]
    impl ManualConnectorHost for TestHost {
        async fn local_addr_for_remote(
            &self,
            remote_addr: SocketAddr,
            _context: SocketContext,
        ) -> anyhow::Result<SocketAddr> {
            Ok(match remote_addr {
                SocketAddr::V4(_) => "127.0.0.1:0".parse().unwrap(),
                SocketAddr::V6(_) => "[::1]:0".parse().unwrap(),
            })
        }

        async fn interface_addrs(&self) -> anyhow::Result<ManualInterfaceAddrs> {
            Ok(ManualInterfaceAddrs {
                interface_ipv4s: vec![],
                interface_ipv6s: vec![],
                public_ipv6: None,
            })
        }
    }

    #[async_trait]
    impl DirectConnectorHost for TestHost {
        async fn collect_ip_addrs(&self, _context: &SocketContext) -> GetIpListResponse {
            GetIpListResponse::default()
        }

        fn mapped_listeners(&self) -> Vec<Url> {
            Vec::new()
        }

        fn is_local_ip(&self, _ip: &IpAddr) -> bool {
            false
        }

        async fn preferred_ipv6_source(
            &self,
            _ip: Ipv6Addr,
            _context: SocketContext,
        ) -> Option<PreferredIpv6Source> {
            None
        }
    }

    fn test_config(network_name: &str) -> CoreInstanceConfig {
        let mut core = CoreConfig::default();
        core.node.network_name = network_name.to_owned();
        core.peer_policy.encryption_required = false;
        let peer = PortablePeerManagerConfig::new(PeerRuntimeConfig {
            core,
            network_identity: NetworkIdentity {
                network_name: network_name.to_owned(),
                network_secret: Some(String::new()),
                network_secret_digest: None,
            },
            stun_info: StunInfo::default(),
            feature_flags: Default::default(),
            secure_mode: None,
            host_routing: HostRoutingPolicy::default(),
        });
        let connectivity = CoreConnectivityConfig::default();
        CoreInstanceConfig {
            instance_name: network_name.to_owned(),
            peer,
            connectivity,
        }
    }

    fn runtime_snapshot(config: &CoreInstanceConfig) -> CoreInstanceRuntimeConfig {
        CoreInstanceRuntimeConfig {
            services: config.connectivity.runtime.clone(),
            peer: Arc::new(config.peer.snapshot.clone()),
        }
    }

    fn proxy_network(real: &str, mapped: Option<&str>) -> ProxyNetworkConfig {
        fn prefix(value: &str) -> IpPrefix {
            let (address, prefix_len) = value.split_once('/').unwrap();
            IpPrefix {
                address: address.parse().unwrap(),
                prefix_len: prefix_len.parse().unwrap(),
            }
        }

        ProxyNetworkConfig {
            real: prefix(real),
            mapped: mapped.map(prefix),
        }
    }

    fn adapters(
        external_listener_factory: Option<
            Arc<dyn ExternalListenerFactory<AcceptedTransport<TestTcpSocket>>>,
        >,
        packet_sink: Arc<dyn PacketSink>,
    ) -> CoreHostAdapters<TestHost> {
        adapters_with_process_runtime(
            external_listener_factory,
            packet_sink,
            CoreProcessRuntime::new(),
        )
    }

    fn adapters_with_process_runtime(
        external_listener_factory: Option<
            Arc<dyn ExternalListenerFactory<AcceptedTransport<TestTcpSocket>>>,
        >,
        packet_sink: Arc<dyn PacketSink>,
        process_runtime: Arc<CoreProcessRuntime>,
    ) -> CoreHostAdapters<TestHost> {
        adapters_with_host_and_process_runtime(
            Arc::new(TestHost::default()),
            external_listener_factory,
            packet_sink,
            process_runtime,
        )
    }

    #[cfg(any(feature = "proxy-packet", feature = "proxy-smoltcp-stack"))]
    fn adapters_with_host(
        host: Arc<TestHost>,
        external_listener_factory: Option<
            Arc<dyn ExternalListenerFactory<AcceptedTransport<TestTcpSocket>>>,
        >,
        packet_sink: Arc<dyn PacketSink>,
    ) -> CoreHostAdapters<TestHost> {
        adapters_with_host_and_process_runtime(
            host,
            external_listener_factory,
            packet_sink,
            CoreProcessRuntime::new(),
        )
    }

    fn adapters_with_host_and_process_runtime(
        host: Arc<TestHost>,
        external_listener_factory: Option<
            Arc<dyn ExternalListenerFactory<AcceptedTransport<TestTcpSocket>>>,
        >,
        packet_sink: Arc<dyn PacketSink>,
        process_runtime: Arc<CoreProcessRuntime>,
    ) -> CoreHostAdapters<TestHost> {
        let dns = Arc::new(TestDns);
        let mut adapters = CoreHostAdapters::new(host, dns, packet_sink, process_runtime);
        adapters.external_listener_factory = external_listener_factory;
        adapters
    }

    fn build_with_engines(
        config: CoreInstanceConfig,
        engines: WrappedTransportEngines,
    ) -> anyhow::Result<Arc<CoreInstance<TestHost>>> {
        build_with_engines_and_listener(config, engines, None)
    }

    fn build_with_engines_and_listener(
        config: CoreInstanceConfig,
        engines: WrappedTransportEngines,
        external_listener_factory: Option<
            Arc<dyn ExternalListenerFactory<AcceptedTransport<TestTcpSocket>>>,
        >,
    ) -> anyhow::Result<Arc<CoreInstance<TestHost>>> {
        let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16);
        let mut adapters = adapters(external_listener_factory, Arc::new(packet_sink));
        adapters.wrapped_transports = engines;
        CoreInstance::new(config, adapters)
    }

    fn build_instance(config: CoreInstanceConfig) -> anyhow::Result<Arc<CoreInstance<TestHost>>> {
        build_with_engines(config, WrappedTransportEngines::default())
    }

    #[cfg(feature = "dhcp-ipv4")]
    #[tokio::test]
    async fn runtime_update_preserves_dhcp_owned_ipv4() {
        let mut initial = test_config("dhcp-runtime-update");
        initial.connectivity.runtime.dhcp_ipv4 = true;
        let instance = build_instance(initial).unwrap();
        let lease = IpPrefix {
            address: "10.126.126.7".parse().unwrap(),
            prefix_len: 24,
        };
        instance.runtime_config.update_peer_with(|peer| {
            peer.runtime.core.routes.ipv4 = Some(lease.clone());
        });

        let mut replacement = test_config("dhcp-runtime-update");
        replacement.connectivity.runtime.dhcp_ipv4 = true;
        instance
            .update_runtime_config(runtime_snapshot(&replacement))
            .await
            .unwrap();

        assert_eq!(
            instance
                .runtime_config
                .snapshot()
                .peer
                .runtime
                .core
                .routes
                .ipv4,
            Some(lease)
        );

        let static_replacement = test_config("dhcp-runtime-update");
        instance
            .update_runtime_config(runtime_snapshot(&static_replacement))
            .await
            .unwrap();
        assert_eq!(
            instance
                .runtime_config
                .snapshot()
                .peer
                .runtime
                .core
                .routes
                .ipv4,
            None
        );
    }

    #[tokio::test]
    async fn core_instance_is_a_direct_managed_record() {
        let instance = build_instance(test_config("managed-directly")).unwrap();

        assert_eq!(instance.instance_name(), "managed-directly");
        assert_eq!(
            instance.instance_id(),
            crate::instance::manager::ManagedInstance::instance_id(instance.as_ref())
        );
    }

    #[tokio::test]
    async fn instance_start_ignores_configured_peer_id() {
        let mut config = test_config("fresh-peer-id");
        let instance_id = uuid::Uuid::from_bytes([7; 16]);
        config.peer.snapshot.runtime.core.node.instance_id = Some(*instance_id.as_bytes());
        config.peer.snapshot.runtime.core.node.peer_id = Some(0);

        let instance = build_instance(config).unwrap();
        let generated_peer_id = instance.peer_id();

        assert_eq!(instance.instance_id(), instance_id);
        assert_ne!(generated_peer_id, 0);
        assert_eq!(
            instance
                .runtime_config
                .snapshot()
                .peer
                .runtime
                .core
                .node
                .peer_id,
            Some(generated_peer_id)
        );
    }

    #[cfg(all(feature = "proxy-packet", feature = "management"))]
    #[tokio::test]
    async fn process_management_rpc_resolves_and_calls_core_instance_directly() {
        use crate::{
            config::toml::TomlConfig,
            instance::manager::{InstanceFactory, InstanceManager},
            management::InstanceManagementRpc,
        };
        use easytier_proto::{
            api::config::{ConfigRpc, GetConfigRequest, InstanceConfigPatch, PatchConfigRequest},
            api::instance::{
                PeerManageRpc, ShowNodeInfoRequest,
                instance_identifier::{InstanceSelector, Selector},
            },
            rpc_types::controller::BaseController,
        };

        struct ManagementTestFactory;

        impl InstanceFactory for ManagementTestFactory {
            type Instance = CoreInstance<TestHost>;
            type CreateContext = ();
            type Error = anyhow::Error;

            fn create(
                &self,
                config: TomlConfig,
                (): Self::CreateContext,
            ) -> Result<Arc<Self::Instance>, Self::Error> {
                let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16);
                let mut adapters = adapters(None, Arc::new(packet_sink));
                adapters.config.force_exit_node = true;
                adapters.config.public_ipv6_provider_supported = true;
                adapters.config.easytier_version = "host-version".to_owned();
                CoreInstance::from_toml(config, adapters)
            }
        }

        let manager = Arc::new(InstanceManager::new(ManagementTestFactory, None));
        let config = TomlConfig::new_from_str(
            r#"
instance_name = "managed-by-name"
hostname = "core-owned-config"
"#,
        )
        .unwrap();
        let instance = manager.create(config, ()).unwrap();
        instance.start().await.unwrap();
        let rpc = InstanceManagementRpc::<ManagementTestFactory>::new(manager);

        let response = rpc
            .show_node_info(
                BaseController::default(),
                ShowNodeInfoRequest {
                    instance: Some(easytier_proto::api::instance::InstanceIdentifier {
                        selector: Some(Selector::InstanceSelector(InstanceSelector {
                            name: Some("managed-by-name".to_owned()),
                        })),
                    }),
                },
            )
            .await
            .unwrap();

        let node = response.node_info.unwrap();
        assert_eq!(node.hostname, "core-owned-config");
        assert!(node.config.contains("instance_name = \"managed-by-name\""));

        let selector = || easytier_proto::api::instance::InstanceIdentifier {
            selector: Some(Selector::InstanceSelector(InstanceSelector {
                name: Some("managed-by-name".to_owned()),
            })),
        };
        rpc.patch_config(
            BaseController::default(),
            PatchConfigRequest {
                patch: Some(InstanceConfigPatch {
                    hostname: Some("patched-in-core".to_owned()),
                    ..Default::default()
                }),
                instance: Some(selector()),
            },
        )
        .await
        .unwrap();
        let response = rpc
            .get_config(
                BaseController::default(),
                GetConfigRequest {
                    instance: Some(selector()),
                },
            )
            .await
            .unwrap();
        assert_eq!(
            response.config.unwrap().hostname.as_deref(),
            Some("patched-in-core")
        );
        let runtime = instance.runtime_config.snapshot();
        assert!(runtime.services.proxy.enable_exit_node);
        assert!(runtime.services.public_ipv6_provider.provider_supported);
        assert_eq!(runtime.peer.easytier_version, "host-version");
    }

    #[cfg(feature = "management")]
    #[tokio::test]
    async fn config_patch_rejects_instance_before_host_startup_is_complete() {
        let instance = build_instance(test_config("not-ready-for-patch")).unwrap();

        let error = crate::management::apply_config_patch(
            &instance,
            easytier_proto::api::config::InstanceConfigPatch {
                hostname: Some("too-early".to_owned()),
                ..Default::default()
            },
        )
        .await
        .unwrap_err();

        assert!(error.to_string().contains("instance is not ready"));
    }

    #[cfg(all(feature = "management", not(feature = "proxy-smoltcp-stack")))]
    #[tokio::test]
    async fn unavailable_gateway_patch_does_not_commit_shared_toml() {
        use easytier_proto::{
            api::config::{ConfigPatchAction, InstanceConfigPatch, PortForwardPatch},
            common::{PortForwardConfigPb, SocketType},
        };

        let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16);
        let instance = CoreInstance::from_toml(
            crate::config::toml::TomlConfig::new_from_str(
                "instance_name = \"rejected-gateway-patch\"",
            )
            .unwrap(),
            adapters(None, Arc::new(packet_sink)),
        )
        .unwrap();
        instance.start().await.unwrap();

        let error = crate::management::apply_config_patch(
            &instance,
            InstanceConfigPatch {
                port_forwards: vec![PortForwardPatch {
                    action: ConfigPatchAction::Add as i32,
                    cfg: Some(PortForwardConfigPb {
                        bind_addr: Some(
                            "127.0.0.1:18080"
                                .parse::<std::net::SocketAddr>()
                                .unwrap()
                                .into(),
                        ),
                        dst_addr: Some(
                            "10.144.144.2:8080"
                                .parse::<std::net::SocketAddr>()
                                .unwrap()
                                .into(),
                        ),
                        socket_type: SocketType::Tcp as i32,
                    }),
                }],
                ..Default::default()
            },
        )
        .await
        .unwrap_err();

        assert!(
            error
                .to_string()
                .contains("does not include the smoltcp gateway"),
            "unexpected patch error: {error:#}"
        );
        assert!(
            instance
                .toml_config()
                .unwrap()
                .get_port_forwards()
                .is_empty()
        );
        assert!(
            instance
                .runtime_config
                .snapshot()
                .services
                .gateway
                .port_forwards
                .is_empty()
        );
        instance.stop().await;
    }

    #[cfg(feature = "web-client")]
    #[tokio::test]
    async fn ignored_gateway_patch_remains_in_toml_config() {
        use easytier_proto::{
            api::config::{ConfigPatchAction, InstanceConfigPatch, PortForwardPatch, UrlPatch},
            common::{PortForwardConfigPb, SocketType},
        };

        let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16);
        let toml_config = crate::config::toml::TomlConfig::new_from_str(
            "instance_name = \"ignored-gateway-patch\"",
        )
        .unwrap();
        let mut host_adapters = adapters(None, Arc::new(packet_sink));
        host_adapters.config.ignore_unsupported_config = true;
        host_adapters.config.gateway_enabled = false;
        host_adapters.config.endpoint_protocols = vec!["tcp".to_owned(), "udp".to_owned()];
        let instance = CoreInstance::from_toml(toml_config, host_adapters).unwrap();
        instance.start().await.unwrap();

        crate::management::apply_config_patch(
            &instance,
            InstanceConfigPatch {
                port_forwards: vec![PortForwardPatch {
                    action: ConfigPatchAction::Add as i32,
                    cfg: Some(PortForwardConfigPb {
                        bind_addr: Some(
                            "127.0.0.1:18080"
                                .parse::<std::net::SocketAddr>()
                                .unwrap()
                                .into(),
                        ),
                        dst_addr: Some(
                            "10.144.144.2:8080"
                                .parse::<std::net::SocketAddr>()
                                .unwrap()
                                .into(),
                        ),
                        socket_type: SocketType::Tcp as i32,
                    }),
                }],
                connectors: vec![UrlPatch {
                    action: ConfigPatchAction::Add as i32,
                    url: Some("quic://127.0.0.1:11010".parse::<url::Url>().unwrap().into()),
                }],
                ..Default::default()
            },
        )
        .await
        .unwrap();

        assert_eq!(instance.toml_config().unwrap().get_port_forwards().len(), 1);
        assert!(
            instance
                .runtime_config
                .snapshot()
                .services
                .gateway
                .port_forwards
                .is_empty()
        );
        assert!(instance.list_connectors().is_empty());
        instance.stop().await;
    }

    #[tokio::test]
    async fn dropping_core_instance_requests_host_shutdown() {
        struct DropAwareRuntimeHost(Arc<AtomicBool>);

        #[async_trait]
        impl InstanceRuntimeHost for DropAwareRuntimeHost {
            async fn prepare(
                &self,
                _packet_plane: Arc<CorePacketPlane>,
            ) -> anyhow::Result<Option<Arc<dyn DhcpIpv4Host>>> {
                Ok(None)
            }

            async fn shutdown(&self) {}

            fn request_shutdown(&self) {
                self.0.store(true, Ordering::Release);
            }
        }

        let shutdown_requested = Arc::new(AtomicBool::new(false));
        let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16);
        let mut adapters = adapters(None, Arc::new(packet_sink));
        adapters.instance_runtime = Arc::new(DropAwareRuntimeHost(shutdown_requested.clone()));
        let instance = CoreInstance::new(test_config("drop-cleanup"), adapters).unwrap();

        drop(instance);

        assert!(shutdown_requested.load(Ordering::Acquire));
    }

    #[tokio::test]
    async fn host_prepare_failure_runs_unified_cleanup() {
        #[derive(Default)]
        struct FailingRuntimeHost {
            shutdown_calls: AtomicUsize,
        }

        #[async_trait]
        impl InstanceRuntimeHost for FailingRuntimeHost {
            async fn prepare(
                &self,
                _packet_plane: Arc<CorePacketPlane>,
            ) -> anyhow::Result<Option<Arc<dyn DhcpIpv4Host>>> {
                anyhow::bail!("host prepare failed")
            }

            async fn shutdown(&self) {
                self.shutdown_calls.fetch_add(1, Ordering::Relaxed);
            }
        }

        let runtime_host = Arc::new(FailingRuntimeHost::default());
        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();
        let instance = CoreInstance::new(test_config("host-prepare-failure"), adapters).unwrap();

        let error = instance.start().await.unwrap_err();

        assert!(error.to_string().contains("host prepare failed"));
        assert_eq!(instance.state(), CoreInstanceState::Stopped);
        assert!(!instance.is_ready());
        assert_eq!(runtime_host.shutdown_calls.load(Ordering::Relaxed), 1);
        assert!(
            instance
                .latest_error()
                .unwrap()
                .contains("host prepare failed")
        );
    }

    #[tokio::test]
    async fn aborting_host_prepare_runs_unified_cleanup() {
        #[derive(Default)]
        struct BlockingPrepareRuntimeHost {
            prepare_started: Notify,
            prepare_release: Notify,
            shutdown_calls: AtomicUsize,
        }

        #[async_trait]
        impl InstanceRuntimeHost for BlockingPrepareRuntimeHost {
            async fn prepare(
                &self,
                _packet_plane: Arc<CorePacketPlane>,
            ) -> anyhow::Result<Option<Arc<dyn DhcpIpv4Host>>> {
                self.prepare_started.notify_one();
                self.prepare_release.notified().await;
                Ok(None)
            }

            async fn shutdown(&self) {
                self.shutdown_calls.fetch_add(1, Ordering::Relaxed);
            }
        }

        let runtime_host = Arc::new(BlockingPrepareRuntimeHost::default());
        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();
        let instance = CoreInstance::new(test_config("aborted-host-prepare"), adapters).unwrap();
        let start = tokio::spawn({
            let instance = instance.clone();
            async move { instance.start().await }
        });
        runtime_host.prepare_started.notified().await;

        start.abort();
        assert!(start.await.unwrap_err().is_cancelled());
        tokio::time::timeout(Duration::from_secs(2), async {
            while instance.state() != CoreInstanceState::Stopped {
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("aborted Host prepare should stop the instance");

        assert!(!instance.is_ready());
        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() {
        use std::{path::Path, path::PathBuf};

        use crate::{
            config::toml::TomlConfig,
            instance::manager::InstanceFactory,
            management::{
                ConfigFileControl, ConfigFilePermission, ConfigFileStorage, InstanceManager,
                InstanceMutationHooks, ProcessManagementRpc,
            },
        };
        use easytier_proto::{
            api::manage::{DeleteNetworkInstanceRequest, WebClientService},
            rpc_types::controller::BaseController,
        };

        #[derive(Default)]
        struct BlockingRuntimeHost {
            shutdown_started: Notify,
            shutdown_release: Notify,
        }

        #[async_trait]
        impl InstanceRuntimeHost for BlockingRuntimeHost {
            async fn prepare(
                &self,
                _packet_plane: Arc<CorePacketPlane>,
            ) -> anyhow::Result<Option<Arc<dyn DhcpIpv4Host>>> {
                Ok(None)
            }

            async fn shutdown(&self) {
                self.shutdown_started.notify_one();
                self.shutdown_release.notified().await;
            }
        }

        struct BlockingFactory {
            runtime_host: Arc<BlockingRuntimeHost>,
            process_runtime: Arc<CoreProcessRuntime>,
        }

        impl InstanceFactory for BlockingFactory {
            type Instance = CoreInstance<TestHost>;
            type CreateContext = ();
            type Error = anyhow::Error;

            fn create(
                &self,
                config: TomlConfig,
                (): Self::CreateContext,
            ) -> Result<Arc<Self::Instance>, Self::Error> {
                let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16);
                let mut adapters = adapters_with_process_runtime(
                    None,
                    Arc::new(packet_sink),
                    self.process_runtime.clone(),
                );
                adapters.instance_runtime = self.runtime_host.clone();
                CoreInstance::from_toml(config, adapters)
            }
        }

        impl crate::management::ProcessRuntimeProvider for BlockingFactory {
            fn process_runtime(&self) -> Arc<CoreProcessRuntime> {
                self.process_runtime.clone()
            }
        }

        #[derive(Default)]
        struct RecordingStorage(AtomicBool);

        #[async_trait]
        impl ConfigFileStorage for RecordingStorage {
            async fn inspect(&self, path: &Path) -> ConfigFileControl {
                ConfigFileControl::new(Some(path.to_owned()), ConfigFilePermission::default())
            }

            async fn read(&self, _path: &Path) -> anyhow::Result<Option<Vec<u8>>> {
                Ok(None)
            }

            async fn write(&self, _path: &Path, _contents: &[u8]) -> anyhow::Result<()> {
                Ok(())
            }

            async fn remove(&self, _path: &Path) -> anyhow::Result<()> {
                self.0.store(true, Ordering::Release);
                Ok(())
            }
        }

        #[derive(Default)]
        struct RecordingHooks(AtomicBool);

        #[async_trait]
        impl InstanceMutationHooks for RecordingHooks {
            async fn post_remove_network_instances(
                &self,
                _instance_ids: &[uuid::Uuid],
            ) -> Result<(), String> {
                self.0.store(true, Ordering::Release);
                Ok(())
            }
        }

        let runtime_host = Arc::new(BlockingRuntimeHost::default());
        let process_runtime = CoreProcessRuntime::new();
        let instances = Arc::new(InstanceManager::new(
            BlockingFactory {
                runtime_host: runtime_host.clone(),
                process_runtime,
            },
            Some(tokio::runtime::Handle::current()),
        ));
        let config = TomlConfig::default();
        config.set_listeners(Vec::new());
        let instance_id = config.get_id();
        instances
            .run_network_instance(
                config,
                ConfigFileControl::new(
                    Some(PathBuf::from("cancelled-delete.toml")),
                    ConfigFilePermission::default(),
                ),
            )
            .unwrap();
        let storage = Arc::new(RecordingStorage::default());
        let hooks = Arc::new(RecordingHooks::default());
        let rpc = ProcessManagementRpc::<BlockingFactory>::new(
            instances.clone(),
            hooks.clone(),
            storage.clone(),
        );

        let deletion = tokio::spawn(async move {
            rpc.delete_network_instance(
                BaseController::default(),
                DeleteNetworkInstanceRequest {
                    inst_ids: vec![instance_id.into()],
                },
            )
            .await
        });
        runtime_host.shutdown_started.notified().await;
        deletion.abort();
        assert!(deletion.await.unwrap_err().is_cancelled());

        let mut wait = tokio::spawn({
            let instances = instances.clone();
            async move { instances.wait().await }
        });
        assert!(
            tokio::time::timeout(Duration::from_millis(20), &mut wait)
                .await
                .is_err()
        );

        runtime_host.shutdown_release.notify_one();
        tokio::time::timeout(Duration::from_secs(1), async {
            while !storage.0.load(Ordering::Acquire) {
                tokio::task::yield_now().await;
            }
        })
        .await
        .unwrap();
        tokio::time::timeout(Duration::from_secs(1), wait)
            .await
            .unwrap()
            .unwrap();
        assert!(hooks.0.load(Ordering::Acquire));
        assert!(instances.instances().is_empty());
    }

    #[cfg(feature = "web-client")]
    #[tokio::test]
    async fn process_management_rpc_owns_instance_create_list_and_delete() {
        use crate::{
            config::toml::TomlConfig,
            instance::manager::InstanceFactory,
            management::{
                InstanceManager, ProcessManagementRpc, UnsupportedConfigFileStorage,
                register_web_client_rpc,
            },
            rpc::service_registry::ServiceRegistry,
        };
        use easytier_proto::{
            api::manage::{
                DeleteNetworkInstanceRequest, ListNetworkInstanceRequest, NetworkConfig,
                NetworkingMethod, RunNetworkInstanceRequest, WebClientService,
            },
            common::RpcDescriptor,
            rpc_types::controller::BaseController,
        };

        struct ManagementTestFactory(Arc<CoreProcessRuntime>);

        impl InstanceFactory for ManagementTestFactory {
            type Instance = CoreInstance<TestHost>;
            type CreateContext = ();
            type Error = anyhow::Error;

            fn create(
                &self,
                config: TomlConfig,
                (): Self::CreateContext,
            ) -> Result<Arc<Self::Instance>, Self::Error> {
                let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16);
                CoreInstance::from_toml(
                    config,
                    adapters_with_process_runtime(None, Arc::new(packet_sink), self.0.clone()),
                )
            }
        }

        impl crate::management::ProcessRuntimeProvider for ManagementTestFactory {
            fn process_runtime(&self) -> Arc<CoreProcessRuntime> {
                self.0.clone()
            }
        }

        let instances = Arc::new(InstanceManager::new(
            ManagementTestFactory(CoreProcessRuntime::new()),
            Some(tokio::runtime::Handle::current()),
        ));
        let registry = ServiceRegistry::new();
        register_web_client_rpc(
            instances.clone(),
            &registry,
            Arc::new(()),
            Arc::new(UnsupportedConfigFileStorage),
        );
        assert_eq!(
            registry.get_method_name(&RpcDescriptor {
                domain_name: String::new(),
                service_name: "ConfigRpc".to_owned(),
                proto_name: "ConfigRpc".to_owned(),
                method_index: 2,
            }),
            Some("get_config".to_owned())
        );
        assert_eq!(
            registry.get_method_name(&RpcDescriptor {
                domain_name: String::new(),
                service_name: "WebClientService".to_owned(),
                proto_name: "WebClientService".to_owned(),
                method_index: 1,
            }),
            Some("validate_config".to_owned())
        );
        let rpc = ProcessManagementRpc::<ManagementTestFactory>::new(
            instances.clone(),
            Arc::new(()),
            Arc::new(UnsupportedConfigFileStorage),
        );
        let created = rpc
            .run_network_instance(
                BaseController::default(),
                RunNetworkInstanceRequest {
                    config: Some(NetworkConfig {
                        network_name: Some("managed-process-rpc".to_owned()),
                        networking_method: Some(NetworkingMethod::Standalone.into()),
                        ..Default::default()
                    }),
                    overwrite: true,
                    ..Default::default()
                },
            )
            .await
            .unwrap()
            .inst_id
            .unwrap();

        let listed = rpc
            .list_network_instance(BaseController::default(), ListNetworkInstanceRequest {})
            .await
            .unwrap();
        assert_eq!(listed.inst_ids, vec![created]);

        let deleted = rpc
            .delete_network_instance(
                BaseController::default(),
                DeleteNetworkInstanceRequest {
                    inst_ids: vec![created],
                },
            )
            .await
            .unwrap();
        assert!(deleted.remain_inst_ids.is_empty());
        assert!(instances.instances().is_empty());
    }

    #[cfg(feature = "management")]
    #[tokio::test]
    async fn owned_selection_and_cleanup_share_the_canonical_transaction() {
        use crate::{
            config::toml::TomlConfig,
            instance::manager::InstanceFactory,
            management::{
                ConfigFileControl, InstanceManager, InstanceMutationHooks, ProcessManagement,
                UnsupportedConfigFileStorage,
            },
        };

        struct ManagementTestFactory(Arc<CoreProcessRuntime>);

        impl InstanceFactory for ManagementTestFactory {
            type Instance = CoreInstance<TestHost>;
            type CreateContext = ();
            type Error = anyhow::Error;

            fn create(
                &self,
                config: TomlConfig,
                (): Self::CreateContext,
            ) -> Result<Arc<Self::Instance>, Self::Error> {
                let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16);
                CoreInstance::from_toml(
                    config,
                    adapters_with_process_runtime(None, Arc::new(packet_sink), self.0.clone()),
                )
            }
        }

        impl crate::management::ProcessRuntimeProvider for ManagementTestFactory {
            fn process_runtime(&self) -> Arc<CoreProcessRuntime> {
                self.0.clone()
            }
        }

        #[derive(Default)]
        struct BlockingRemovalHook {
            entered: Notify,
            release: Notify,
            removed: std::sync::Mutex<Vec<Vec<uuid::Uuid>>>,
        }

        #[async_trait]
        impl InstanceMutationHooks for BlockingRemovalHook {
            async fn post_remove_network_instances(
                &self,
                instance_ids: &[uuid::Uuid],
            ) -> Result<(), String> {
                self.removed.lock().unwrap().push(instance_ids.to_vec());
                self.entered.notify_one();
                self.release.notified().await;
                Ok(())
            }
        }

        let instances = Arc::new(InstanceManager::new(
            ManagementTestFactory(CoreProcessRuntime::new()),
            Some(tokio::runtime::Handle::current()),
        ));
        let config = TomlConfig::default();
        config.set_listeners(Vec::new());
        let instance_id = config.get_id();
        instances
            .run_network_instance(config, ConfigFileControl::STATIC_CONFIG)
            .unwrap();
        let hooks = Arc::new(BlockingRemovalHook::default());
        let management = ProcessManagement::<ManagementTestFactory>::new(
            instances.clone(),
            hooks.clone(),
            Arc::new(UnsupportedConfigFileStorage),
        );

        let deletion_management = management.clone();
        let deletion = tokio::spawn(async move {
            deletion_management
                .delete_owned_network_instances(vec![instance_id, uuid::Uuid::new_v4()])
                .await
        });
        hooks.entered.notified().await;
        assert!(instances.mutation_lock().try_lock().is_err());
        assert_eq!(
            hooks.removed.lock().unwrap().as_slice(),
            &[vec![instance_id]]
        );
        hooks.release.notify_one();

        let result = deletion.await.unwrap().unwrap();
        assert_eq!(result.removed_instance_ids, vec![instance_id]);
        assert_eq!(hooks.removed.lock().unwrap().len(), 1);

        let old_name = format!("old-{instance_id}");
        let new_name = format!("new-{instance_id}");
        let old_config = TomlConfig::default();
        old_config.set_id(instance_id);
        old_config.set_inst_name(old_name.clone());
        old_config.set_listeners(Vec::new());
        instances
            .run_network_instance(old_config, ConfigFileControl::STATIC_CONFIG)
            .unwrap();

        let mutation_guard = instances.mutation_lock().lock_owned().await;
        let deletion = management.delete_owned_network_instances_by_name(vec![old_name]);
        tokio::pin!(deletion);
        assert!(matches!(
            futures::poll!(deletion.as_mut()),
            std::task::Poll::Pending
        ));

        instances
            .delete_network_instances([instance_id])
            .await
            .unwrap();
        let new_config = TomlConfig::default();
        new_config.set_id(instance_id);
        new_config.set_inst_name(new_name.clone());
        new_config.set_listeners(Vec::new());
        instances
            .run_network_instance(new_config, ConfigFileControl::STATIC_CONFIG)
            .unwrap();

        drop(mutation_guard);
        hooks.release.notify_one();
        let result = deletion.await.unwrap();
        assert!(result.removed_instance_ids.is_empty());
        assert_eq!(
            instances
                .instance(instance_id)
                .map(|instance| instance.instance_name().to_owned())
                .as_deref(),
            Some(new_name.as_str())
        );
        assert_eq!(
            hooks.removed.lock().unwrap().as_slice(),
            &[vec![instance_id], Vec::new()]
        );
        instances
            .delete_network_instances([instance_id])
            .await
            .unwrap();

        let selection_ran = Arc::new(std::sync::atomic::AtomicBool::new(false));
        let selection_ran_for_call = selection_ran.clone();
        let mutation_guard = instances.mutation_lock().lock_owned().await;
        let deletion = management.delete_owned_network_instances_selected_by(move || {
            selection_ran_for_call.store(true, std::sync::atomic::Ordering::Release);
            Vec::new()
        });
        tokio::pin!(deletion);
        assert!(matches!(
            futures::poll!(deletion.as_mut()),
            std::task::Poll::Pending
        ));
        assert!(!selection_ran.load(std::sync::atomic::Ordering::Acquire));
        drop(mutation_guard);
        hooks.release.notify_one();
        deletion.await.unwrap();
        assert!(selection_ran.load(std::sync::atomic::Ordering::Acquire));
    }

    #[cfg(feature = "management")]
    #[tokio::test]
    async fn process_management_rpc_rolls_back_instance_and_file_on_hook_failure() {
        use std::{
            collections::HashMap,
            path::{Path, PathBuf},
            sync::Mutex as StdMutex,
        };

        use crate::{
            config::toml::TomlConfig,
            instance::manager::InstanceFactory,
            management::{
                ConfigFileControl, ConfigFilePermission, ConfigFileStorage, InstanceManager,
                InstanceMutationHooks, ProcessManagementRpc,
            },
        };
        use easytier_proto::{
            api::manage::{
                NetworkConfig, NetworkingMethod, RunNetworkInstanceRequest, WebClientService,
            },
            rpc_types::controller::BaseController,
        };

        struct ManagementTestFactory(Arc<CoreProcessRuntime>);

        impl InstanceFactory for ManagementTestFactory {
            type Instance = CoreInstance<TestHost>;
            type CreateContext = ();
            type Error = anyhow::Error;

            fn create(
                &self,
                config: TomlConfig,
                (): Self::CreateContext,
            ) -> Result<Arc<Self::Instance>, Self::Error> {
                let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16);
                CoreInstance::from_toml(
                    config,
                    adapters_with_process_runtime(None, Arc::new(packet_sink), self.0.clone()),
                )
            }
        }

        impl crate::management::ProcessRuntimeProvider for ManagementTestFactory {
            fn process_runtime(&self) -> Arc<CoreProcessRuntime> {
                self.0.clone()
            }
        }

        #[derive(Default)]
        struct MemoryStorage(StdMutex<HashMap<PathBuf, Vec<u8>>>);

        #[async_trait]
        impl ConfigFileStorage for MemoryStorage {
            async fn inspect(&self, path: &Path) -> ConfigFileControl {
                ConfigFileControl::new(Some(path.to_owned()), ConfigFilePermission::default())
            }

            async fn read(&self, path: &Path) -> anyhow::Result<Option<Vec<u8>>> {
                Ok(self.0.lock().unwrap().get(path).cloned())
            }

            async fn write(&self, path: &Path, contents: &[u8]) -> anyhow::Result<()> {
                self.0
                    .lock()
                    .unwrap()
                    .insert(path.to_owned(), contents.to_vec());
                Ok(())
            }

            async fn remove(&self, path: &Path) -> anyhow::Result<()> {
                self.0.lock().unwrap().remove(path);
                Ok(())
            }
        }

        struct RejectPostRun;

        #[async_trait]
        impl InstanceMutationHooks for RejectPostRun {
            fn manages_remote_config_instances(&self) -> bool {
                true
            }

            async fn post_run_network_instance(
                &self,
                _instance_id: &uuid::Uuid,
            ) -> Result<(), String> {
                Err("rejected for rollback test".to_owned())
            }
        }

        let instances = Arc::new(InstanceManager::new(
            ManagementTestFactory(CoreProcessRuntime::new()),
            Some(tokio::runtime::Handle::current()),
        ));
        let storage = Arc::new(MemoryStorage::default());
        let instance_id = uuid::Uuid::new_v4();
        let config_path = PathBuf::from("managed-process-rpc.toml");
        let original_file = b"original configuration".to_vec();
        storage
            .0
            .lock()
            .unwrap()
            .insert(config_path.clone(), original_file.clone());
        let original = TomlConfig::default();
        original.set_id(instance_id);
        original.set_inst_name("original-instance".to_owned());
        original.set_listeners(Vec::new());
        instances
            .run_network_instance(
                original,
                ConfigFileControl::new(Some(config_path.clone()), ConfigFilePermission::default()),
            )
            .unwrap();
        let rpc = ProcessManagementRpc::<ManagementTestFactory>::new(
            instances.clone(),
            Arc::new(RejectPostRun),
            storage.clone(),
        );

        let result = rpc
            .run_network_instance(
                BaseController::default(),
                RunNetworkInstanceRequest {
                    inst_id: Some(instance_id.into()),
                    config: Some(NetworkConfig {
                        network_name: Some("replacement".to_owned()),
                        networking_method: Some(NetworkingMethod::Standalone.into()),
                        ..Default::default()
                    }),
                    overwrite: true,
                    ..Default::default()
                },
            )
            .await;

        assert!(result.is_err());
        assert_eq!(
            instances
                .instance(instance_id)
                .map(|instance| instance.instance_name().to_owned())
                .as_deref(),
            Some("original-instance")
        );
        assert_eq!(
            storage.0.lock().unwrap().get(&config_path),
            Some(&original_file)
        );
        instances
            .delete_network_instances([instance_id])
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn packet_plane_does_not_retain_core_instance() {
        let instance = build_instance(test_config("packet-plane-ownership")).unwrap();
        let weak = Arc::downgrade(&instance);
        let packet_plane = instance.packet_plane();

        drop(instance);

        assert!(weak.upgrade().is_none());
        drop(packet_plane);
    }

    #[derive(Default)]
    struct RecordingProxyService {
        start_calls: AtomicUsize,
        stop_calls: AtomicUsize,
        start_gate: Option<Arc<ProxyStartGate>>,
        #[cfg(feature = "proxy-packet")]
        destination_ingress: StdMutex<
            Option<crate::gateway::proxy::wrapped_transport::WrappedTransportDestinationIngress>,
        >,
    }

    #[derive(Default)]
    struct ProxyStartGate {
        entered: Notify,
        release: Notify,
    }

    impl RecordingProxyService {
        fn blocking() -> (Arc<Self>, Arc<ProxyStartGate>) {
            let gate = Arc::new(ProxyStartGate::default());
            (
                Arc::new(Self {
                    start_gate: Some(gate.clone()),
                    ..Default::default()
                }),
                gate,
            )
        }

        #[cfg(feature = "proxy-packet")]
        fn destination_ingress(
            &self,
        ) -> Option<crate::gateway::proxy::wrapped_transport::WrappedTransportDestinationIngress>
        {
            self.destination_ingress.lock().unwrap().clone()
        }
    }

    #[async_trait]
    impl WrappedTransportEngine for RecordingProxyService {
        async fn prepare(&self, options: WrappedTransportEngineStart) -> anyhow::Result<()> {
            self.start_calls.fetch_add(1, Ordering::Relaxed);
            #[cfg(feature = "proxy-packet")]
            {
                *self.destination_ingress.lock().unwrap() = options.destination_ingress;
            }
            #[cfg(not(feature = "proxy-packet"))]
            let _ = options;
            if let Some(gate) = &self.start_gate {
                gate.entered.notify_one();
                gate.release.notified().await;
            }
            Ok(())
        }

        async fn activate(&self) -> anyhow::Result<()> {
            Ok(())
        }

        async fn inject_peer_datagram(
            &self,
            _role: WrappedTransportRole,
            _from_peer_id: u32,
            _payload: bytes::Bytes,
        ) -> anyhow::Result<()> {
            Ok(())
        }

        #[cfg(feature = "proxy-packet")]
        async fn connect_source(
            &self,
            _request: crate::gateway::proxy::wrapped_transport::WrappedTransportConnect,
        ) -> anyhow::Result<Box<dyn crate::gateway::proxy::traits::TcpProxyStream>> {
            anyhow::bail!("recording engine does not open streams")
        }

        async fn stop(&self) {
            self.stop_calls.fetch_add(1, Ordering::Relaxed);
        }
    }

    #[derive(Debug, Default)]
    struct BlockingListenerState {
        start_entered: Notify,
        drop_calls: AtomicUsize,
    }

    #[derive(Debug)]
    struct BlockingSocketListener {
        url: Url,
        state: Arc<BlockingListenerState>,
    }

    #[async_trait]
    impl SocketListener for BlockingSocketListener {
        type Accepted = AcceptedTransport<TestTcpSocket>;

        async fn listen(&mut self) -> anyhow::Result<()> {
            self.state.start_entered.notify_one();
            std::future::pending().await
        }

        async fn accept(&mut self) -> anyhow::Result<Self::Accepted> {
            std::future::pending().await
        }

        fn local_url(&self) -> Url {
            self.url.clone()
        }
    }

    impl Drop for BlockingSocketListener {
        fn drop(&mut self) {
            self.state.drop_calls.fetch_add(1, Ordering::Relaxed);
        }
    }

    struct BlockingExternalListenerFactory {
        state: Arc<BlockingListenerState>,
    }

    impl ExternalListenerFactory<AcceptedTransport<TestTcpSocket>> for BlockingExternalListenerFactory {
        fn supports_scheme(&self, scheme: &str) -> bool {
            scheme == "unix"
        }

        fn create(
            &self,
            request: ExternalListenerRequest,
        ) -> Box<dyn SocketListener<Accepted = AcceptedTransport<TestTcpSocket>>> {
            Box::new(BlockingSocketListener {
                url: request.url,
                state: self.state.clone(),
            })
        }
    }

    #[derive(Debug)]
    struct ReadySocketListener(Url);

    #[async_trait]
    impl SocketListener for ReadySocketListener {
        type Accepted = AcceptedTransport<TestTcpSocket>;

        async fn listen(&mut self) -> anyhow::Result<()> {
            Ok(())
        }

        async fn accept(&mut self) -> anyhow::Result<Self::Accepted> {
            std::future::pending().await
        }

        fn local_url(&self) -> Url {
            self.0.clone()
        }
    }

    struct ReadyExternalListenerFactory;

    impl ExternalListenerFactory<AcceptedTransport<TestTcpSocket>> for ReadyExternalListenerFactory {
        fn supports_scheme(&self, scheme: &str) -> bool {
            scheme == "unix"
        }

        fn create(
            &self,
            request: ExternalListenerRequest,
        ) -> Box<dyn SocketListener<Accepted = AcceptedTransport<TestTcpSocket>>> {
            Box::new(ReadySocketListener(request.url))
        }
    }

    #[tokio::test]
    async fn runtime_updates_refresh_avoid_relay_preference() {
        let config = test_config("portable-runtime-update");
        let instance = build_instance(config.clone()).unwrap();

        assert!(
            !instance
                .node_snapshot()
                .await
                .feature_flags
                .avoid_relay_data
        );

        let mut enabled = runtime_snapshot(&config);
        Arc::make_mut(&mut enabled.peer).avoid_relay_data_preference = true;
        instance.update_runtime_config(enabled).await.unwrap();
        assert!(
            instance
                .node_snapshot()
                .await
                .feature_flags
                .avoid_relay_data
        );

        let mut disabled = runtime_snapshot(&config);
        Arc::make_mut(&mut disabled.peer).avoid_relay_data_preference = false;
        instance.update_runtime_config(disabled).await.unwrap();
        assert!(
            !instance
                .node_snapshot()
                .await
                .feature_flags
                .avoid_relay_data
        );
    }

    #[cfg(all(feature = "test-utils", feature = "dhcp-ipv4"))]
    #[cfg_attr(
        not(target_os = "wasi"),
        tokio::test(flavor = "multi_thread", worker_threads = 2)
    )]
    #[cfg_attr(target_os = "wasi", tokio::test)]
    async fn concurrent_runtime_updates_keep_snapshot_and_derived_state_coherent() {
        let config = test_config("concurrent-runtime-update");
        let instance = build_instance(config.clone()).unwrap();
        instance.start().await.unwrap();

        let original = instance.node_snapshot().await;
        let mut full = runtime_snapshot(&config);
        full.services.dhcp_ipv4 = true;
        full.services.acl.tcp_whitelist = vec!["80".to_owned()];
        {
            let peer = Arc::make_mut(&mut full.peer);
            peer.runtime.core.node.hostname = Some("full".to_owned());
            peer.runtime.core.routes.proxy_networks =
                vec![proxy_network("192.0.2.0/24", Some("198.51.100.0/24"))];
        }
        let mut peer_update = full.clone();
        {
            let peer = Arc::make_mut(&mut peer_update.peer);
            peer.runtime.core.node.hostname = Some("peer".to_owned());
            peer.runtime.core.routes.proxy_networks =
                vec![proxy_network("203.0.113.0/24", Some("10.20.30.0/24"))];
        }

        let start = Arc::new(tokio::sync::Barrier::new(3));
        let full_update = tokio::spawn({
            let instance = instance.clone();
            let start = start.clone();
            async move {
                start.wait().await;
                instance.update_runtime_config(full).await
            }
        });
        let peer_update = tokio::spawn({
            let instance = instance.clone();
            let start = start.clone();
            async move {
                start.wait().await;
                instance.update_runtime_config(peer_update).await
            }
        });
        start.wait().await;
        full_update.await.unwrap().unwrap();
        peer_update.await.unwrap().unwrap();

        let final_config = instance.runtime_config.snapshot();
        assert!(final_config.services.dhcp_ipv4);
        assert_eq!(instance.acl_whitelist_snapshot().tcp_ports, ["80"]);
        assert_eq!(instance.acl_reload_count.load(Ordering::Relaxed), 1);
        let node = instance.node_snapshot().await;
        assert_eq!(node.peer_id, original.peer_id);
        assert_eq!(node.instance_id, original.instance_id);
        assert_eq!(
            final_config.peer.runtime.core.node.hostname.as_deref(),
            Some(node.hostname.as_str())
        );
        match node.hostname.as_str() {
            "full" => assert_eq!(
                instance
                    .proxy_cidr_table
                    .lookup_v4("198.51.100.42".parse().unwrap()),
                Some("192.0.2.42".parse().unwrap())
            ),
            "peer" => assert_eq!(
                instance
                    .proxy_cidr_table
                    .lookup_v4("10.20.30.42".parse().unwrap()),
                Some("203.0.113.42".parse().unwrap())
            ),
            hostname => panic!("unexpected final hostname: {hostname:?}"),
        }

        instance.stop().await;
    }

    #[cfg(all(feature = "test-utils", feature = "dhcp-ipv4"))]
    #[tokio::test]
    async fn active_runtime_update_skips_unchanged_and_rejects_invalid_acl() {
        let config = test_config("invalid-active-acl-update");
        let instance = build_instance(config.clone()).unwrap();
        instance.start().await.unwrap();

        let mut unrelated = runtime_snapshot(&config);
        Arc::make_mut(&mut unrelated.peer)
            .runtime
            .core
            .node
            .hostname = Some("accepted".to_owned());
        instance.update_runtime_config(unrelated).await.unwrap();
        assert_eq!(instance.acl_reload_count.load(Ordering::Relaxed), 0);
        let before = instance.node_snapshot().await;

        let mut rejected = runtime_snapshot(&config);
        rejected.services.dhcp_ipv4 = true;
        rejected.services.acl.tcp_whitelist = vec!["invalid".to_owned()];
        Arc::make_mut(&mut rejected.peer).runtime.core.node.hostname = Some("rejected".to_owned());

        let error = instance.update_runtime_config(rejected).await.unwrap_err();
        assert!(error.to_string().contains("Invalid port number"));
        assert!(!instance.runtime_config.snapshot().services.dhcp_ipv4);
        assert!(instance.acl_whitelist_snapshot().tcp_ports.is_empty());
        assert_eq!(instance.acl_reload_count.load(Ordering::Relaxed), 0);
        assert_eq!(instance.node_snapshot().await.hostname, before.hostname);
        instance.stop().await;
    }

    #[cfg(feature = "proxy-packet")]
    #[tokio::test]
    async fn runtime_core_instance_owns_connectivity_lifecycle() {
        let mut config = test_config("connectivity-lifecycle");
        config.peer.snapshot.runtime.core.routes.proxy_networks =
            vec![proxy_network("10.1.2.0/24", None)];
        let initial_peer: Url = "tcp://127.0.0.1:29999".parse().unwrap();
        config.connectivity.initial_peers = vec![initial_peer.clone()];
        let proxy = Arc::new(RecordingProxyService::default());
        let instance = build_with_engines(
            config,
            WrappedTransportEngines {
                kcp: Some(proxy.clone()),
                quic: None,
            },
        )
        .unwrap();

        assert_eq!(instance.state(), CoreInstanceState::Created);
        assert_eq!(instance.list_connectors().len(), 1);
        assert_eq!(instance.list_connectors()[0].url, initial_peer);
        instance.start().await.unwrap();
        assert_eq!(instance.state(), CoreInstanceState::Running);
        assert!(instance.is_ready());
        assert!(instance.start().await.is_err());
        assert_eq!(proxy.start_calls.load(Ordering::Relaxed), 1);

        instance.stop().await;
        instance.stop().await;
        assert_eq!(instance.state(), CoreInstanceState::Stopped);
        assert_eq!(proxy.stop_calls.load(Ordering::Relaxed), 1);
    }

    #[cfg(feature = "proxy-smoltcp-stack")]
    #[tokio::test]
    async fn startup_plan_controls_gateway_for_initial_and_updated_config() {
        fn build(config: CoreInstanceConfig) -> (Arc<CoreInstance<TestHost>>, Arc<TestHost>) {
            let host = Arc::new(TestHost {
                reject_socks5_listener: true,
                ..Default::default()
            });
            let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16);
            let adapters = adapters_with_host(host.clone(), None, Arc::new(packet_sink));
            (CoreInstance::new(config, adapters).unwrap(), host)
        }

        let mut enabled_config = test_config("gateway-enabled-by-default");
        enabled_config.connectivity.runtime.gateway.socks5_bind =
            Some("127.0.0.1:1080".parse().unwrap());
        let (enabled, _) = build(enabled_config);
        let error = enabled.start().await.unwrap_err();
        assert!(error.to_string().contains("rejected SOCKS5 listener"));
        assert_eq!(enabled.state(), CoreInstanceState::Stopped);

        let mut disabled_config = test_config("gateway-disabled-by-plan");
        disabled_config.connectivity.startup_plan.gateway = false;
        disabled_config.peer.snapshot.runtime.core.routes.ipv4 =
            Some(IpPrefix::new("10.144.0.1".parse().unwrap(), 24).unwrap());
        let (disabled, _) = build(disabled_config.clone());
        let mut updated = runtime_snapshot(&disabled_config);
        updated.services.gateway.socks5_bind = Some("127.0.0.1:1080".parse().unwrap());
        disabled.update_runtime_config(updated).await.unwrap();
        disabled.start().await.unwrap();
        assert_eq!(disabled.state(), CoreInstanceState::Running);
        let socket = disabled
            .data_plane_udp_bind(0, Duration::from_secs(1))
            .await
            .expect("startup plan must not disable the data-plane runtime");
        drop(socket);
        disabled.stop().await;
    }

    #[cfg(feature = "proxy-smoltcp-stack")]
    #[tokio::test]
    async fn failed_port_forward_start_releases_started_listeners() {
        let mut config = test_config("port-forward-start-rollback");
        config.connectivity.runtime.gateway.port_forwards = vec![
            PortForwardConfig {
                bind_addr: "127.0.0.1:18080".parse().unwrap(),
                dst_addr: "10.144.0.2:80".parse().unwrap(),
                proto: "tcp".to_owned(),
            },
            PortForwardConfig {
                bind_addr: "127.0.0.1:18081".parse().unwrap(),
                dst_addr: "10.144.0.2:81".parse().unwrap(),
                proto: "unsupported".to_owned(),
            },
        ];
        let host = Arc::new(TestHost::default());
        let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16);
        let adapters = adapters_with_host(host.clone(), None, Arc::new(packet_sink));
        let instance = CoreInstance::new(config, adapters).unwrap();

        let error = instance.start().await.unwrap_err();

        assert!(error.to_string().contains("unsupported protocol"));
        assert_eq!(instance.state(), CoreInstanceState::Stopped);
        assert_eq!(host.active_tcp_listeners.load(Ordering::Relaxed), 0);
    }

    #[cfg(feature = "proxy-packet")]
    #[tokio::test]
    async fn runtime_core_instance_owns_wrapped_transport_source_nat() {
        let mut config = test_config("wrapped-source");
        config.peer.snapshot.flags.enable_kcp_proxy = true;
        config.peer.snapshot.flags.disable_kcp_input = true;
        let engine = Arc::new(RecordingProxyService::default());
        let instance = build_with_engines(
            config,
            WrappedTransportEngines {
                kcp: Some(engine.clone()),
                quic: None,
            },
        )
        .unwrap();

        instance.start().await.unwrap();
        assert!(
            instance.wrapped_transport_is_started(
                WrappedTransportKind::Kcp,
                WrappedTransportRole::Source,
            )
        );
        assert!(
            instance
                .wrapped_tcp_proxy_entry_snapshots(
                    WrappedTransportKind::Kcp,
                    WrappedTransportRole::Source,
                )
                .is_empty()
        );

        instance.stop().await;
        assert!(
            !instance.wrapped_transport_is_started(
                WrappedTransportKind::Kcp,
                WrappedTransportRole::Source,
            )
        );
        assert_eq!(engine.stop_calls.load(Ordering::Relaxed), 1);
    }

    #[cfg(feature = "proxy-packet")]
    #[tokio::test]
    async fn runtime_core_instance_owns_wrapped_transport_destination_sessions() {
        let mut config = test_config("wrapped-destination");
        config.peer.snapshot.flags.enable_kcp_proxy = false;
        config.peer.snapshot.flags.disable_kcp_input = false;
        let engine = Arc::new(RecordingProxyService::default());
        let (connections, mut connection_receiver) = tokio::sync::mpsc::unbounded_channel();
        let host = Arc::new(TestHost {
            proxy_nat_connections: Some(connections),
            ..Default::default()
        });
        let (packet_sink, _packet_receiver) = tokio::sync::mpsc::channel(16);
        let mut adapters = adapters_with_host(host, None, Arc::new(packet_sink));
        adapters.wrapped_transports = WrappedTransportEngines {
            kcp: Some(engine.clone()),
            quic: None,
        };
        let instance = CoreInstance::new(config, adapters).unwrap();

        instance.start().await.unwrap();
        assert!(instance.wrapped_transport_is_started(
            WrappedTransportKind::Kcp,
            WrappedTransportRole::Destination,
        ));

        let destination: SocketAddr = "127.0.0.1:20100".parse().unwrap();
        let ingress = engine
            .destination_ingress()
            .expect("core should inject a destination ingress");
        let (core_stream, peer_stream) = tokio::io::duplex(1024);
        ingress
            .submit(
                crate::gateway::proxy::wrapped_transport::WrappedTransportAcceptedStream {
                    src: "10.0.0.2:40000".parse().unwrap(),
                    dst: destination,
                    initial_acl_packet_size: 16,
                    stream: Box::new(core_stream),
                },
            )
            .await
            .unwrap();
        let (connected_destination, destination_stream) = tokio::time::timeout(
            std::time::Duration::from_secs(2),
            connection_receiver.recv(),
        )
        .await
        .expect("core should request the destination socket")
        .unwrap();
        assert_eq!(connected_destination, destination);
        tokio::time::timeout(std::time::Duration::from_secs(2), async {
            loop {
                let entries = instance.wrapped_tcp_proxy_entry_snapshots(
                    WrappedTransportKind::Kcp,
                    WrappedTransportRole::Destination,
                );
                if entries.iter().any(|entry| {
                    entry.state
                        == crate::gateway::proxy::tcp_proxy_engine::TcpNatEntryState::Connected
                }) {
                    break;
                }
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("core should own the connected destination entry");

        drop(peer_stream);
        drop(destination_stream);
        tokio::time::timeout(std::time::Duration::from_secs(2), async {
            while !instance
                .wrapped_tcp_proxy_entry_snapshots(
                    WrappedTransportKind::Kcp,
                    WrappedTransportRole::Destination,
                )
                .is_empty()
            {
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("completed destination entry should be removed");

        let (core_stream, _blocked_peer_stream) = tokio::io::duplex(1024);
        ingress
            .submit(
                crate::gateway::proxy::wrapped_transport::WrappedTransportAcceptedStream {
                    src: "10.0.0.2:40001".parse().unwrap(),
                    dst: destination,
                    initial_acl_packet_size: 16,
                    stream: Box::new(core_stream),
                },
            )
            .await
            .unwrap();
        let (connected_destination, _blocked_destination_stream) = tokio::time::timeout(
            std::time::Duration::from_secs(2),
            connection_receiver.recv(),
        )
        .await
        .expect("second destination session should request a socket")
        .unwrap();
        assert_eq!(connected_destination, destination);
        tokio::time::timeout(std::time::Duration::from_secs(2), async {
            while instance
                .wrapped_tcp_proxy_entry_snapshots(
                    WrappedTransportKind::Kcp,
                    WrappedTransportRole::Destination,
                )
                .is_empty()
            {
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("blocked destination session should be visible");

        tokio::time::timeout(std::time::Duration::from_secs(2), instance.stop())
            .await
            .expect("stop should cancel core-owned destination sessions");
        assert!(
            instance
                .wrapped_tcp_proxy_entry_snapshots(
                    WrappedTransportKind::Kcp,
                    WrappedTransportRole::Destination,
                )
                .is_empty()
        );
        assert!(
            ingress
                .submit(
                    crate::gateway::proxy::wrapped_transport::WrappedTransportAcceptedStream {
                        src: "10.0.0.2:40002".parse().unwrap(),
                        dst: destination,
                        initial_acl_packet_size: 16,
                        stream: Box::new(tokio::io::duplex(64).0),
                    },
                )
                .await
                .is_err()
        );
    }

    #[tokio::test]
    async fn runtime_core_instance_owns_the_transport_proxy_cidr_table() {
        let mut config = test_config("transport-proxy-cidr");
        config.connectivity.runtime.proxy.forward_by_system = true;
        config.peer.snapshot.runtime.core.routes.proxy_networks =
            vec![proxy_network("192.0.2.0/24", Some("198.51.100.0/24"))];
        let mut updated = runtime_snapshot(&config);
        let proxy = Arc::new(RecordingProxyService::default());
        let instance = build_with_engines(
            config,
            WrappedTransportEngines {
                kcp: Some(proxy.clone()),
                quic: None,
            },
        )
        .unwrap();

        assert_eq!(instance.node_snapshot().await.proxy_networks.len(), 1);
        instance.start().await.unwrap();
        assert_eq!(proxy.start_calls.load(Ordering::Relaxed), 1);

        Arc::make_mut(&mut updated.peer)
            .runtime
            .core
            .routes
            .proxy_networks = vec![proxy_network("203.0.113.0/24", Some("10.20.30.0/24"))];
        instance.update_runtime_config(updated).await.unwrap();
        let proxy_networks = instance.node_snapshot().await.proxy_networks;
        assert_eq!(proxy_networks.len(), 1);
        assert_eq!(
            proxy_networks[0].real.address,
            "203.0.113.0".parse::<IpAddr>().unwrap()
        );
        assert_eq!(
            proxy_networks[0].mapped.as_ref().unwrap().address,
            "10.20.30.0".parse::<IpAddr>().unwrap()
        );

        instance.stop().await;
        assert!(instance.start().await.is_err());
        assert_eq!(proxy.stop_calls.load(Ordering::Relaxed), 1);
    }

    #[tokio::test]
    async fn runtime_core_rejects_invalid_acl_runtime_snapshot() {
        let config = test_config("explicit-acl");
        let instance = build_instance(config.clone()).unwrap();
        assert_eq!(instance.acl_whitelist_snapshot(), Default::default());

        let mut updated = runtime_snapshot(&config);
        updated.services.acl.tcp_whitelist = vec!["invalid".to_owned()];
        let error = instance.update_runtime_config(updated).await.unwrap_err();
        assert!(error.to_string().contains("Invalid port number"));
        assert!(instance.acl_whitelist_snapshot().tcp_ports.is_empty());
        instance.start().await.unwrap();
        instance.stop().await;
    }

    #[cfg(feature = "dhcp-ipv4")]
    #[tokio::test]
    async fn runtime_core_accepts_explicit_dhcp_runtime_snapshot() {
        let config = test_config("explicit-dhcp");
        let instance = build_instance(config.clone()).unwrap();
        let mut updated = runtime_snapshot(&config);
        updated.services.dhcp_ipv4 = true;
        instance.update_runtime_config(updated).await.unwrap();
        let error = instance.start().await.unwrap_err();
        assert!(error.to_string().contains("no host adapter was provided"));
        assert_eq!(instance.state(), CoreInstanceState::Stopped);
    }

    #[cfg(feature = "public-ipv6-provider")]
    #[tokio::test]
    async fn runtime_core_accepts_explicit_public_ipv6_runtime_snapshot() {
        let config = test_config("explicit-public-ipv6");
        let instance = build_instance(config.clone()).unwrap();
        let mut updated = runtime_snapshot(&config);
        updated.services.public_ipv6_provider.provider_enabled = true;
        updated.services.public_ipv6_provider.provider_supported = true;
        updated.services.public_ipv6_provider.configured_prefix =
            Some("fd00::/64".parse().unwrap());
        instance.update_runtime_config(updated).await.unwrap();

        let error = instance.start().await.unwrap_err();
        assert!(error.to_string().contains("not a valid global unicast"));
        assert_eq!(instance.state(), CoreInstanceState::Stopped);
    }

    #[cfg(not(feature = "proxy-packet"))]
    #[test]
    fn runtime_core_rejects_packet_proxy_requests_when_unavailable() {
        let mut config = test_config("packet-proxy-unavailable");
        config.connectivity.runtime.proxy.enable_exit_node = true;

        let error = match build_instance(config) {
            Ok(_) => panic!("packet proxy request unexpectedly succeeded"),
            Err(error) => error,
        };

        assert!(
            error
                .to_string()
                .contains("does not include packet proxy services")
        );
    }

    #[cfg(not(feature = "proxy-smoltcp-stack"))]
    #[tokio::test]
    async fn runtime_core_rejects_unavailable_gateway_updates() {
        let config = test_config("smoltcp-gateway-update-unavailable");
        let instance = build_instance(config.clone()).unwrap();
        let mut updated = runtime_snapshot(&config);
        updated.services.gateway.socks5_bind = Some("127.0.0.1:1080".parse().unwrap());

        let error = instance.update_runtime_config(updated).await.unwrap_err();

        assert!(
            error
                .to_string()
                .contains("does not include the smoltcp gateway")
        );
    }

    #[tokio::test]
    async fn stopping_while_transport_proxy_starts_rolls_back_once() {
        let mut config = test_config("blocking-transport-proxy");
        config.connectivity.runtime.proxy.forward_by_system = true;
        config.peer.snapshot.runtime.core.routes.proxy_networks =
            vec![proxy_network("10.1.2.0/24", None)];
        config.connectivity.initial_peers = vec!["tcp://127.0.0.1:29998".parse().unwrap()];
        let (proxy, start_gate) = RecordingProxyService::blocking();
        let instance = build_with_engines(
            config,
            WrappedTransportEngines {
                kcp: Some(proxy.clone()),
                quic: None,
            },
        )
        .unwrap();

        let start_task = tokio::spawn({
            let instance = instance.clone();
            async move { instance.start().await }
        });
        start_gate.entered.notified().await;
        let stop_task = tokio::spawn({
            let instance = instance.clone();
            async move { instance.stop().await }
        });
        while !instance.cancel.is_cancelled() {
            tokio::task::yield_now().await;
        }

        assert!(start_task.await.unwrap().is_err());
        stop_task.await.unwrap();
        assert_eq!(instance.list_connectors().len(), 1);
        assert_eq!(instance.state(), CoreInstanceState::Stopped);
        assert!(!instance.is_ready());
        assert_eq!(proxy.start_calls.load(Ordering::Relaxed), 1);
        assert_eq!(proxy.stop_calls.load(Ordering::Relaxed), 1);
    }

    #[tokio::test]
    async fn start_serializes_runtime_updates() {
        let mut config = test_config("serialized-start");
        config.connectivity.runtime.proxy.forward_by_system = true;
        config.peer.snapshot.runtime.core.routes.proxy_networks =
            vec![proxy_network("10.1.4.0/24", None)];
        let mut updated = runtime_snapshot(&config);
        Arc::make_mut(&mut updated.peer).runtime.core.node.hostname =
            Some("updated-after-start".to_owned());
        let (proxy, start_gate) = RecordingProxyService::blocking();
        let instance = build_with_engines(
            config,
            WrappedTransportEngines {
                kcp: Some(proxy),
                quic: None,
            },
        )
        .unwrap();

        let start = tokio::spawn({
            let instance = instance.clone();
            async move { instance.start().await }
        });
        start_gate.entered.notified().await;

        let update = instance.update_runtime_config(updated);
        tokio::pin!(update);
        assert!(matches!(
            futures::poll!(update.as_mut()),
            std::task::Poll::Pending
        ));

        start_gate.release.notify_one();
        start.await.unwrap().unwrap();
        update.await.unwrap();
        assert_eq!(
            instance.node_snapshot().await.hostname,
            "updated-after-start"
        );
        instance.stop().await;
    }

    #[tokio::test]
    async fn aborting_start_stops_partial_runtime() {
        let mut config = test_config("aborted-start");
        config.connectivity.runtime.proxy.forward_by_system = true;
        config.peer.snapshot.runtime.core.routes.proxy_networks =
            vec![proxy_network("10.1.3.0/24", None)];
        let (proxy, start_gate) = RecordingProxyService::blocking();
        let instance = build_with_engines(
            config,
            WrappedTransportEngines {
                kcp: Some(proxy.clone()),
                quic: None,
            },
        )
        .unwrap();

        let start = tokio::spawn({
            let instance = instance.clone();
            async move { instance.start().await }
        });
        start_gate.entered.notified().await;
        start.abort();
        assert!(start.await.unwrap_err().is_cancelled());

        tokio::time::timeout(Duration::from_secs(2), async {
            while instance.state() != CoreInstanceState::Stopped {
                tokio::task::yield_now().await;
            }
        })
        .await
        .expect("aborted activation should recover the instance");
        assert!(!instance.is_ready());
        assert_eq!(proxy.start_calls.load(Ordering::Relaxed), 1);
        assert_eq!(proxy.stop_calls.load(Ordering::Relaxed), 1);
    }

    #[tokio::test]
    async fn invalid_initial_peer_fails_during_construction() {
        let mut config = test_config("invalid-initial-peer");
        config.connectivity.initial_peers =
            vec!["unsupported://peer.example:1234".parse().unwrap()];

        let error = build_instance(config)
            .err()
            .expect("invalid initial peer should fail construction");
        assert!(
            error
                .to_string()
                .contains("unsupported core manual connector URL"),
            "unexpected construction error: {error:#}"
        );
    }

    #[tokio::test]
    async fn runtime_core_instances_keep_lifecycle_and_connectors_isolated() {
        let instance_a = build_instance(test_config("instance-a")).unwrap();
        let instance_b = build_instance(test_config("instance-b")).unwrap();
        let connector_a: Url = "tcp://127.0.0.1:21001".parse().unwrap();
        let connector_b: Url = "udp://127.0.0.1:21002".parse().unwrap();

        instance_a.add_connector(connector_a.clone()).unwrap();
        instance_b.add_connector(connector_b.clone()).unwrap();
        assert_eq!(instance_a.list_connectors()[0].url, connector_a);
        assert_eq!(instance_b.list_connectors()[0].url, connector_b);
        instance_a.clear_connectors();
        instance_b.clear_connectors();

        let (start_a, start_b) = tokio::join!(instance_a.start(), instance_b.start());
        start_a.unwrap();
        start_b.unwrap();
        assert_eq!(instance_a.state(), CoreInstanceState::Running);
        assert_eq!(instance_b.state(), CoreInstanceState::Running);

        instance_a.stop().await;
        assert_eq!(instance_a.state(), CoreInstanceState::Stopped);
        assert_eq!(instance_b.state(), CoreInstanceState::Running);
        instance_b.stop().await;
        assert_eq!(instance_b.state(), CoreInstanceState::Stopped);
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn stop_cancels_pending_listener_start() {
        let state = Arc::new(BlockingListenerState::default());
        let mut config = test_config("pending-listener");
        config.connectivity.listeners = Some(ListenerRuntimeConfig::new(
            vec!["unix:///tmp/easytier-pending-listener".parse().unwrap()],
            false,
            SocketContext::default(),
        ));
        let instance = build_with_engines_and_listener(
            config,
            WrappedTransportEngines::default(),
            Some(Arc::new(BlockingExternalListenerFactory {
                state: state.clone(),
            })),
        )
        .unwrap();
        let start_instance = instance.clone();
        let start_task =
            AbortOnDropHandle::new(tokio::spawn(async move { start_instance.start().await }));
        let start_result = tokio::time::timeout(Duration::from_secs(1), async {
            state.start_entered.notified().await;
            instance.stop().await;
            start_task.await.unwrap()
        })
        .await
        .expect("listener cancellation should complete promptly");

        assert!(start_result.is_err());
        assert_eq!(instance.state(), CoreInstanceState::Stopped);
        assert_eq!(state.drop_calls.load(Ordering::Relaxed), 1);
    }

    #[tokio::test]
    async fn external_listener_uses_core_running_listener_registry() {
        let external_url: Url = "unix:///tmp/easytier-external-listener-test"
            .parse()
            .unwrap();
        let mut config = test_config("external-listener-registry");
        config.connectivity.listeners = Some(ListenerRuntimeConfig::new(
            vec![external_url.clone()],
            false,
            SocketContext::default(),
        ));
        let instance = build_with_engines_and_listener(
            config,
            WrappedTransportEngines::default(),
            Some(Arc::new(ReadyExternalListenerFactory)),
        )
        .unwrap();

        instance.start().await.unwrap();
        let running = instance.running_listeners();
        assert_eq!(running.len(), 2);
        assert!(running.iter().any(|url| url.scheme() == "ring"));
        assert!(running.contains(&external_url));
        assert_eq!(instance.node_snapshot().await.listeners, running);

        instance.stop().await;
        assert!(instance.running_listeners().is_empty());
    }
}