Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

easytier-web/frontend-lib/src/components/UrlListInput.vue

@@ -11,8 +11,17 @@ const props = defineProps<{
 
 const list = defineModel<string[]>({ required: true })
 
+const fallbackUrl = () => {
+    const protoKeys = Object.keys(props.protos)
+    const defaultProto = protoKeys.includes('tcp')
+        ? 'tcp'
+        : (protoKeys[0] ?? 'tcp')
+    const defaultPort = props.protos[defaultProto] ?? 11010
+    return `${defaultProto}://0.0.0.0:${defaultPort}`
+}
+
 const addUrl = () => {
-    list.value.push(props.defaultUrl || 'tcp://0.0.0.0:11010')
+    list.value.push(props.defaultUrl || fallbackUrl())
 }
 
 const removeUrl = (index: number) => {

easytier-web/frontend-lib/tests/url-list-input.spec.ts

@@ -0,0 +1,93 @@
+import { mount } from '@vue/test-utils'
+import { describe, expect, it } from 'vitest'
+import { defineComponent, h, nextTick, ref } from 'vue'
+import UrlListInput from '../src/components/UrlListInput.vue'
+
+const ButtonStub = defineComponent({
+  name: 'Button',
+  emits: ['click'],
+  setup(_, { slots, emit }) {
+    return () => h('button', { onClick: (event: MouseEvent) => emit('click', event) }, slots.default?.())
+  },
+})
+
+const UrlInputStub = defineComponent({
+  name: 'UrlInput',
+  setup(_, { slots }) {
+    return () => h('div', slots.actions?.())
+  },
+})
+
+function mountUrlListInput(protos: Record<string, number>, defaultUrl?: string) {
+  const urls = ref<string[]>([])
+  const wrapper = mount(defineComponent({
+    components: { UrlListInput },
+    setup() {
+      return { urls, protos, defaultUrl }
+    },
+    template: `
+      <UrlListInput
+        v-model="urls"
+        :protos="protos"
+        :default-url="defaultUrl"
+        add-label="add_url"
+      />
+    `,
+  }), {
+    global: {
+      stubs: {
+        Button: ButtonStub,
+        UrlInput: UrlInputStub,
+      },
+    },
+  })
+
+  return { wrapper, urls }
+}
+
+describe('UrlListInput.vue add fallback', () => {
+  it('derives the fallback URL from protos when defaultUrl is not provided', async () => {
+    const { wrapper, urls } = mountUrlListInput({ tcp: 11010, udp: 11010 })
+
+    await wrapper.find('.cursor-pointer').trigger('click')
+    await nextTick()
+
+    expect(urls.value).toEqual(['tcp://0.0.0.0:11010'])
+  })
+
+  it('falls back to the first available protocol when tcp is not present', async () => {
+    const { wrapper, urls } = mountUrlListInput({ udp: 22000 })
+
+    await wrapper.find('.cursor-pointer').trigger('click')
+    await nextTick()
+
+    expect(urls.value).toEqual(['udp://0.0.0.0:22000'])
+  })
+
+  it('falls back to tcp default port when protos is empty', async () => {
+    const { wrapper, urls } = mountUrlListInput({})
+
+    await wrapper.find('.cursor-pointer').trigger('click')
+    await nextTick()
+
+    expect(urls.value).toEqual(['tcp://0.0.0.0:11010'])
+  })
+
+  it('supports port-zero fallback from protos', async () => {
+    const { wrapper, urls } = mountUrlListInput({ tcp: 0, udp: 0 })
+
+    await wrapper.find('.cursor-pointer').trigger('click')
+    await nextTick()
+
+    expect(urls.value).toEqual(['tcp://0.0.0.0:0'])
+  })
+
+  it('uses defaultUrl when provided', async () => {
+    const { wrapper, urls } = mountUrlListInput({ tcp: 11010 }, 'udp://0.0.0.0:22000')
+
+    await wrapper.find('.cursor-pointer').trigger('click')
+    await nextTick()
+
+    expect(urls.value).toEqual(['udp://0.0.0.0:22000'])
+  })
+})

easytier/Cargo.toml

@@ -32,6 +32,10 @@ path = "src/lib.rs"
 name = "tx_throughput"
 harness = false
 
+[[bench]]
+name = "packet_bytes_extraction"
+harness = false
+
 [dependencies]
 git-version = "0.3.9"
 

easytier/benches/README.md

@@ -1,4 +1,47 @@
-# TX Throughput Benchmark
+# Benchmarks
+
+Criterion benchmarks for EasyTier hot paths.
+
+| Bench                       | What it measures                                                                 |
+| --------------------------- | -------------------------------------------------------------------------------- |
+| `tx_throughput`             | End-to-end TX injection path through `peer_manager::send_msg_by_ip`              |
+| `packet_bytes_extraction`   | `ZCPacket::payload_bytes` / `tunnel_payload_bytes` extraction (advance hot path)  |
+
+## Packet Bytes Extraction
+
+Criterion benchmark for `ZCPacket` bytes extraction — the methods touched by the
+`advance`-based slicing refactor. Measures `payload_bytes` and
+`tunnel_payload_bytes` at two payload sizes (1280, 4096). Setup
+(`ZCPacket::new_with_payload`) runs in the benchmark harness's preparation
+phase and is excluded from the timed region, so the numbers reflect only the
+extraction call.
+
+### Quick start
+
+```bash
+cargo bench --bench packet_bytes_extraction
+```
+
+Smoke run:
+
+```bash
+PACKET_BYTES_MEASUREMENT_SECS=2 \
+PACKET_BYTES_WARMUP_SECS=1 \
+PACKET_BYTES_SAMPLE_SIZE=10 \
+cargo bench --bench packet_bytes_extraction -- --quiet
+```
+
+### Environment variables
+
+| Variable                        | Default | Notes                        |
+| ------------------------------- | ------- | ---------------------------- |
+| `PACKET_BYTES_MEASUREMENT_SECS` | `10`    | Criterion `measurement_time` |
+| `PACKET_BYTES_WARMUP_SECS`      | `3`     | Criterion `warm_up_time`     |
+| `PACKET_BYTES_SAMPLE_SIZE`      | `10`    | Criterion `sample_size` (min 10) |
+
+---
+
+## TX Throughput Benchmark
 
 Criterion benchmark for EasyTier's TX injection path (`peer_manager::send_msg_by_ip`).
 

easytier/benches/packet_bytes_extraction.rs

@@ -0,0 +1,65 @@
+use std::hint::black_box;
+use std::time::Duration;
+
+use criterion::{BatchSize, Criterion, Throughput, criterion_group, criterion_main};
+
+use easytier::tunnel::packet_def::ZCPacket;
+
+const PAYLOAD_SIZES: &[usize] = &[1280, 4096];
+
+fn env_parse<T: std::str::FromStr>(key: &str, default: T) -> T {
+    std::env::var(key)
+        .ok()
+        .and_then(|v| v.parse().ok())
+        .unwrap_or(default)
+}
+
+fn bench_payload_bytes(c: &mut Criterion) {
+    let mut group = c.benchmark_group("payload_bytes");
+    for &size in PAYLOAD_SIZES {
+        let data = vec![0u8; size];
+        group.throughput(Throughput::Bytes(size as u64));
+        group.bench_with_input(format!("{size}"), &data, |b, data| {
+            b.iter_batched(
+                || ZCPacket::new_with_payload(black_box(data)),
+                |p| black_box(p).payload_bytes(),
+                BatchSize::SmallInput,
+            )
+        });
+    }
+    group.finish();
+}
+
+fn bench_tunnel_payload_bytes(c: &mut Criterion) {
+    let mut group = c.benchmark_group("tunnel_payload_bytes");
+    for &size in PAYLOAD_SIZES {
+        let data = vec![0u8; size];
+        group.throughput(Throughput::Bytes(size as u64));
+        group.bench_with_input(format!("{size}"), &data, |b, data| {
+            b.iter_batched(
+                || ZCPacket::new_with_payload(black_box(data)),
+                |p| black_box(p).tunnel_payload_bytes(),
+                BatchSize::SmallInput,
+            )
+        });
+    }
+    group.finish();
+}
+
+fn criterion_config() -> Criterion {
+    let measurement_secs = env_parse("PACKET_BYTES_MEASUREMENT_SECS", 10u64);
+    let warmup_secs = env_parse("PACKET_BYTES_WARMUP_SECS", 3u64);
+    let sample_size = env_parse("PACKET_BYTES_SAMPLE_SIZE", 10usize).max(10);
+
+    Criterion::default()
+        .measurement_time(Duration::from_secs(measurement_secs))
+        .warm_up_time(Duration::from_secs(warmup_secs))
+        .sample_size(sample_size)
+}
+
+criterion_group! {
+    name = benches;
+    config = criterion_config();
+    targets = bench_payload_bytes, bench_tunnel_payload_bytes
+}
+criterion_main!(benches);

easytier/src/common/global_ctx.rs

@@ -219,6 +219,7 @@ pub struct GlobalCtx {
 
     running_listeners: Mutex<Vec<url::Url>>,
     advertised_ipv6_public_addr_prefix: Mutex<Option<cidr::Ipv6Cidr>>,
+    tun_device_name: Mutex<Option<String>>,
 
     flags: ArcSwap<Flags>,
 
@@ -336,6 +337,7 @@ impl GlobalCtx {
 
             running_listeners: Mutex::new(Vec::new()),
             advertised_ipv6_public_addr_prefix: Mutex::new(None),
+            tun_device_name: Mutex::new(None),
 
             flags: ArcSwap::new(Arc::new(flags)),
 
@@ -370,6 +372,24 @@ impl GlobalCtx {
         }
     }
 
+    fn set_tun_device_name(&self, name: Option<String>) {
+        *self.tun_device_name.lock().unwrap() = name;
+    }
+
+    pub(crate) fn set_tun_device_ready(&self, name: String) {
+        self.set_tun_device_name(Some(name.clone()));
+        self.issue_event(GlobalCtxEvent::TunDeviceReady(name));
+    }
+
+    pub(crate) fn set_tun_device_error(&self, error: String) {
+        self.set_tun_device_name(None);
+        self.issue_event(GlobalCtxEvent::TunDeviceError(error));
+    }
+
+    pub fn get_tun_device_name(&self) -> Option<String> {
+        self.tun_device_name.lock().unwrap().clone()
+    }
+
     pub fn check_network_in_whitelist(&self, network_name: &str) -> Result<(), anyhow::Error> {
         if self
             .get_flags()
@@ -825,6 +845,36 @@ pub mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_tun_device_name_tracks_explicit_runtime_state() {
+        let config = TomlConfigLoader::default();
+        let global_ctx = GlobalCtx::new(config);
+
+        assert_eq!(global_ctx.get_tun_device_name(), None);
+
+        global_ctx.issue_event(GlobalCtxEvent::TunDeviceReady("ignored".to_string()));
+        assert_eq!(global_ctx.get_tun_device_name(), None);
+
+        let mut subscriber = global_ctx.subscribe();
+
+        global_ctx.set_tun_device_ready("easytier0".to_string());
+        assert_eq!(
+            global_ctx.get_tun_device_name(),
+            Some("easytier0".to_string())
+        );
+        assert_eq!(
+            subscriber.recv().await.unwrap(),
+            GlobalCtxEvent::TunDeviceReady("easytier0".to_string())
+        );
+
+        global_ctx.set_tun_device_error("closed".to_string());
+        assert_eq!(global_ctx.get_tun_device_name(), None);
+        assert_eq!(
+            subscriber.recv().await.unwrap(),
+            GlobalCtxEvent::TunDeviceError("closed".to_string())
+        );
+    }
+
     #[tokio::test]
     async fn trusted_key_source_lookup_is_precise() {
         let config = TomlConfigLoader::default();

easytier/src/common/ifcfg/mod.rs

@@ -177,3 +177,20 @@ pub(crate) fn list_ipv6_route_messages()
 pub(crate) fn get_interface_index(name: &str) -> Result<u32, Error> {
     netlink::NetlinkIfConfiger::get_interface_index(name)
 }
+
+#[cfg(target_os = "linux")]
+pub(crate) fn add_ipv6_ndp_proxy(name: &str, address: Ipv6Addr) -> Result<(), Error> {
+    netlink::NetlinkIfConfiger::add_ipv6_ndp_proxy(name, address)
+}
+
+#[cfg(target_os = "linux")]
+pub(crate) fn remove_ipv6_ndp_proxy(name: &str, address: Ipv6Addr) -> Result<(), Error> {
+    netlink::NetlinkIfConfiger::remove_ipv6_ndp_proxy(name, address)
+}
+
+#[cfg(target_os = "linux")]
+pub(crate) fn list_ipv6_ndp_proxy(
+    name: &str,
+) -> Result<std::collections::BTreeSet<Ipv6Addr>, Error> {
+    netlink::NetlinkIfConfiger::list_ipv6_ndp_proxy(name)
+}

easytier/src/common/ifcfg/netlink.rs

@@ -1,4 +1,5 @@
 use std::{
+    collections::BTreeSet,
     ffi::CString,
     fmt::Debug,
     net::{IpAddr, Ipv4Addr, Ipv6Addr},
@@ -16,6 +17,10 @@ use netlink_packet_core::{
 use netlink_packet_route::{
     AddressFamily, RouteNetlinkMessage,
     address::{AddressAttribute, AddressMessage},
+    neighbour::{
+        NeighbourAddress, NeighbourAttribute, NeighbourFlags, NeighbourHeader, NeighbourMessage,
+        NeighbourState,
+    },
     route::{
         RouteAddress, RouteAttribute, RouteHeader, RouteMessage, RouteProtocol, RouteScope,
         RouteType,
@@ -375,6 +380,105 @@ impl NetlinkIfConfiger {
     pub(crate) fn list_ipv6_route_messages() -> Result<Vec<RouteMessage>, Error> {
         Self::list_route_messages(AddressFamily::Inet6)
     }
+
+    fn ipv6_ndp_proxy_message(name: &str, address: Ipv6Addr) -> Result<NeighbourMessage, Error> {
+        let mut message = NeighbourMessage::default();
+        message.header = NeighbourHeader {
+            family: AddressFamily::Inet6,
+            ifindex: Self::get_interface_index(name)?,
+            state: NeighbourState::Permanent,
+            flags: NeighbourFlags::Proxy,
+            kind: RouteType::Unicast,
+        };
+        message
+            .attributes
+            .push(NeighbourAttribute::Destination(NeighbourAddress::Inet6(
+                address,
+            )));
+        Ok(message)
+    }
+
+    pub(crate) fn add_ipv6_ndp_proxy(name: &str, address: Ipv6Addr) -> Result<(), Error> {
+        send_netlink_req_and_wait_one_resp(
+            RouteNetlinkMessage::NewNeighbour(Self::ipv6_ndp_proxy_message(name, address)?),
+            false,
+        )
+    }
+
+    pub(crate) fn remove_ipv6_ndp_proxy(name: &str, address: Ipv6Addr) -> Result<(), Error> {
+        send_netlink_req_and_wait_one_resp(
+            RouteNetlinkMessage::DelNeighbour(Self::ipv6_ndp_proxy_message(name, address)?),
+            true,
+        )
+    }
+
+    fn list_neighbour_messages(
+        address_family: AddressFamily,
+    ) -> Result<Vec<NeighbourMessage>, Error> {
+        let mut message = NeighbourMessage::default();
+        message.header.family = address_family;
+        message.header.flags = NeighbourFlags::Proxy;
+
+        let s = send_netlink_req(
+            RouteNetlinkMessage::GetNeighbour(message),
+            NLM_F_REQUEST | NLM_F_DUMP,
+        )?;
+
+        let mut ret_vec = vec![];
+        let mut resp = Vec::<u8>::new();
+        loop {
+            if resp.is_empty() {
+                let (new_resp, _) = s.recv_from_full()?;
+                resp = new_resp;
+            }
+
+            let ret = NetlinkMessage::<RouteNetlinkMessage>::deserialize(&resp)
+                .with_context(|| "Failed to deserialize netlink neighbour message")?;
+            resp = resp.split_off(ret.buffer_len());
+
+            tracing::debug!("net link response <<< {:?}", ret);
+
+            match ret.payload {
+                NetlinkPayload::Error(e) => {
+                    if e.code == NonZero::new(0) {
+                        continue;
+                    } else {
+                        return Err(e.to_io().into());
+                    }
+                }
+                NetlinkPayload::InnerMessage(RouteNetlinkMessage::NewNeighbour(m)) => {
+                    ret_vec.push(m);
+                }
+                NetlinkPayload::Done(_) => {
+                    break;
+                }
+                p => {
+                    tracing::error!("Unexpected netlink response: {:?}", p);
+                    return Err(anyhow::anyhow!("Unexpected netlink response").into());
+                }
+            }
+        }
+
+        Ok(ret_vec)
+    }
+
+    pub(crate) fn list_ipv6_ndp_proxy(name: &str) -> Result<BTreeSet<Ipv6Addr>, Error> {
+        let ifindex = Self::get_interface_index(name)?;
+
+        Ok(Self::list_neighbour_messages(AddressFamily::Inet6)?
+            .into_iter()
+            .filter(|message| {
+                message.header.ifindex == ifindex
+                    && message.header.flags.contains(NeighbourFlags::Proxy)
+            })
+            .filter_map(|message| {
+                message.attributes.into_iter().find_map(|attr| match attr {
+                    NeighbourAttribute::Destination(NeighbourAddress::Inet6(addr)) => Some(addr),
+                    _ => None,
+                })
+            })
+            .collect())
+    }
 }
 
 #[async_trait]

easytier/src/instance/instance.rs

@@ -65,9 +65,9 @@ use crate::vpn_portal::{self, VpnPortal};
 use super::dns_server::{MAGIC_DNS_FAKE_IP, runner::DnsRunner};
 use super::listeners::ListenerManager;
 use super::public_ipv6_provider::{
-    reconcile_public_ipv6_provider_runtime, run_public_ipv6_provider_reconcile_task,
-    should_run_public_ipv6_provider_reconcile, validate_public_ipv6_config,
-    validate_public_ipv6_config_values,
+    PublicIpv6ProviderReconcileTask, reconcile_public_ipv6_provider_runtime,
+    run_public_ipv6_provider_reconcile_task, should_run_public_ipv6_provider_reconcile,
+    validate_public_ipv6_config, validate_public_ipv6_config_values,
 };
 
 #[cfg(feature = "socks5")]
@@ -194,6 +194,44 @@ impl NicCtxContainer {
 
 #[cfg(feature = "tun")]
 type ArcNicCtx = Arc<Mutex<Option<NicCtxContainer>>>;
+type ArcPublicIpv6ProviderTaskSlot = Arc<PublicIpv6ProviderTaskSlot>;
+
+struct PublicIpv6ProviderTaskSlot {
+    task: Mutex<Option<PublicIpv6ProviderReconcileTask>>,
+    closing: AtomicBool,
+}
+
+impl PublicIpv6ProviderTaskSlot {
+    fn new() -> Self {
+        Self {
+            task: Mutex::new(None),
+            closing: AtomicBool::new(false),
+        }
+    }
+
+    async fn ensure_started(&self, global_ctx: &ArcGlobalCtx) {
+        let mut task = self.task.lock().await;
+        if self.closing.load(Ordering::Acquire) || task.is_some() {
+            return;
+        }
+        *task = run_public_ipv6_provider_reconcile_task(global_ctx);
+    }
+
+    async fn shutdown(&self) {
+        self.closing.store(true, Ordering::Release);
+        let task = self.task.lock().await.take();
+        if let Some(task) = task {
+            task.shutdown().await;
+        }
+    }
+}
+
+async fn ensure_public_ipv6_provider_reconcile_task(
+    global_ctx: &ArcGlobalCtx,
+    task_slot: &ArcPublicIpv6ProviderTaskSlot,
+) {
+    task_slot.ensure_started(global_ctx).await;
+}
 
 pub struct InstanceRpcServerHook {
     rpc_portal_whitelist: Vec<IpCidr>,
@@ -254,6 +292,7 @@ pub struct InstanceConfigPatcher {
     socks5_server: Weak<Socks5Server>,
     peer_manager: Weak<PeerManager>,
     conn_manager: Weak<ManualConnectorManager>,
+    public_ipv6_provider_task: ArcPublicIpv6ProviderTaskSlot,
 }
 
 impl InstanceConfigPatcher {
@@ -324,7 +363,6 @@ impl InstanceConfigPatcher {
         self.patch_mapped_listeners(patch.mapped_listeners).await?;
         self.patch_connector(patch.connectors).await?;
 
-        let provider_reconcile_was_running = should_run_public_ipv6_provider_reconcile(&global_ctx);
         let mut provider_config_changed = false;
         if let Some(hostname) = patch.hostname {
             global_ctx.set_hostname(hostname.clone());
@@ -362,10 +400,12 @@ impl InstanceConfigPatcher {
         if provider_config_changed {
             reconcile_public_ipv6_provider_runtime(&global_ctx).await;
 
-            let provider_reconcile_should_run =
-                should_run_public_ipv6_provider_reconcile(&global_ctx);
-            if !provider_reconcile_was_running && provider_reconcile_should_run {
-                run_public_ipv6_provider_reconcile_task(&global_ctx);
+            if should_run_public_ipv6_provider_reconcile(&global_ctx) {
+                ensure_public_ipv6_provider_reconcile_task(
+                    &global_ctx,
+                    &self.public_ipv6_provider_task,
+                )
+                .await;
             }
         }
 
@@ -647,6 +687,7 @@ pub struct Instance {
     socks5_server: Arc<Socks5Server>,
 
     proxy_cidrs_monitor: Option<AbortOnDropHandle<()>>,
+    public_ipv6_provider_task: ArcPublicIpv6ProviderTaskSlot,
 
     global_ctx: ArcGlobalCtx,
 }
@@ -734,6 +775,7 @@ impl Instance {
             socks5_server,
 
             proxy_cidrs_monitor: None,
+            public_ipv6_provider_task: Arc::new(PublicIpv6ProviderTaskSlot::new()),
 
             global_ctx,
         }
@@ -1034,7 +1076,11 @@ impl Instance {
             .await?;
         self.listener_manager.lock().await.run().await?;
         self.peer_manager.run().await?;
-        run_public_ipv6_provider_reconcile_task(&self.global_ctx);
+        ensure_public_ipv6_provider_reconcile_task(
+            &self.global_ctx,
+            &self.public_ipv6_provider_task,
+        )
+        .await;
 
         #[cfg(feature = "tun")]
         {
@@ -1347,6 +1393,7 @@ impl Instance {
             socks5_server: Arc::downgrade(&self.socks5_server),
             peer_manager: Arc::downgrade(&self.peer_manager),
             conn_manager: Arc::downgrade(&self.conn_manager),
+            public_ipv6_provider_task: self.public_ipv6_provider_task.clone(),
         }
     }
 
@@ -1602,6 +1649,7 @@ impl Instance {
     }
 
     pub async fn clear_resources(&mut self) {
+        self.public_ipv6_provider_task.shutdown().await;
         self.peer_manager.clear_resources().await;
         #[cfg(feature = "tun")]
         let _ = self.nic_ctx.lock().await.take();
@@ -1787,6 +1835,21 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn public_ipv6_provider_task_slot_does_not_restart_after_shutdown() {
+        let global_ctx = get_mock_global_ctx();
+        let slot = std::sync::Arc::new(super::PublicIpv6ProviderTaskSlot::new());
+        global_ctx.config.set_ipv6_public_addr_provider(true);
+        global_ctx
+            .config
+            .set_ipv6_public_addr_prefix(Some("2001:db8::/48".parse().unwrap()));
+
+        slot.shutdown().await;
+        super::ensure_public_ipv6_provider_reconcile_task(&global_ctx, &slot).await;
+
+        assert!(slot.task.lock().await.is_none());
+    }
+
     #[tokio::test]
     async fn validate_public_ipv6_patch_allows_enabling_auto_with_manual_ipv6() {
         let global_ctx = get_mock_global_ctx();

easytier/src/instance/virtual_nic.rs

@@ -24,7 +24,7 @@ use crate::{
 };
 
 use byteorder::WriteBytesExt as _;
-use bytes::{BufMut, BytesMut};
+use bytes::{Buf, BufMut, BytesMut};
 use cidr::{Ipv4Inet, Ipv6Inet};
 use futures::{SinkExt, Stream, StreamExt, lock::BiLock, ready};
 use pin_project_lite::pin_project;
@@ -180,12 +180,13 @@ impl ZCPacketToBytes for TunZCPacketToBytes {
         assert!(payload_offset >= 4);
 
         let ret = if self.has_packet_info {
-            let mut inner = inner.split_off(payload_offset - 4);
+            inner.advance(payload_offset - 4);
             let proto = infer_proto(&inner[4..]);
             self.fill_packet_info(&mut inner[0..4], proto)?;
             inner
         } else {
-            inner.split_off(payload_offset)
+            inner.advance(payload_offset);
+            inner
         };
 
         tracing::debug!(?ret, ?payload_offset, "convert zc packet to tun packet");
@@ -1361,12 +1362,11 @@ impl NicCtx {
                     }
 
                     self.global_ctx
-                        .issue_event(GlobalCtxEvent::TunDeviceReady(nic.ifname().to_string()));
+                        .set_tun_device_ready(nic.ifname().to_string());
                     ret
                 }
                 Err(err) => {
-                    self.global_ctx
-                        .issue_event(GlobalCtxEvent::TunDeviceError(err.to_string()));
+                    self.global_ctx.set_tun_device_error(err.to_string());
                     return Err(err);
                 }
             }
@@ -1405,12 +1405,11 @@ impl NicCtx {
             match nic.create_dev_for_mobile(tun_fd).await {
                 Ok(ret) => {
                     self.global_ctx
-                        .issue_event(GlobalCtxEvent::TunDeviceReady(nic.ifname().to_string()));
+                        .set_tun_device_ready(nic.ifname().to_string());
                     ret
                 }
                 Err(err) => {
-                    self.global_ctx
-                        .issue_event(GlobalCtxEvent::TunDeviceError(err.to_string()));
+                    self.global_ctx.set_tun_device_error(err.to_string());
                     return Err(err);
                 }
             }

easytier/src/launcher.rs

@@ -173,7 +173,12 @@ impl EasyTierLauncher {
         #[cfg(mobile)]
         Self::run_routine_for_mobile(&instance, &data, &mut tasks).await;
 
-        instance.run().await?;
+        if let Err(err) = instance.run().await {
+            tasks.abort_all();
+            drop(tasks);
+            instance.clear_resources().await;
+            return Err(err.into());
+        }
 
         #[cfg(feature = "ffi-dataplane")]
         data.data_plane

easytier/src/peers/peer_conn.rs

@@ -363,15 +363,7 @@ impl PeerConn {
         let throughput = peer_conn_tunnel_filter.filter_output();
         let filter_chain = TunnelFilterChain::new(session_filter.clone(), peer_conn_tunnel_filter);
         let peer_conn_tunnel = TunnelWithFilter::new(tunnel, filter_chain);
-        let supports_direct = peer_conn_tunnel
-            .info()
-            .map(|i| matches!(i.tunnel_type.as_str(), "ring" | "udp" | "tcp"))
-            .unwrap_or(false);
-        let mut mpsc_tunnel = if supports_direct {
-            MpscTunnel::new_direct(peer_conn_tunnel)
-        } else {
-            MpscTunnel::new(peer_conn_tunnel, Some(Duration::from_secs(7)))
-        };
+        let mut mpsc_tunnel = MpscTunnel::new(peer_conn_tunnel, Some(Duration::from_secs(7)));
 
         let (recv, sink) = (mpsc_tunnel.get_stream(), mpsc_tunnel.get_sink());
 

easytier/src/tests/three_node.rs

@@ -477,6 +477,12 @@ struct PublicIpv6Lab {
     extra_bridges: [&'static str; 2],
 }
 
+#[derive(Clone, Copy)]
+enum PublicIpv6LabTopology {
+    DelegatedPrefix,
+    OnLinkPrefix,
+}
+
 impl PublicIpv6Lab {
     const PROVIDER_NS: &'static str = "net_a";
     const CLIENT_NS: &'static str = "net_b";
@@ -490,11 +496,13 @@ impl PublicIpv6Lab {
     const PROVIDER_DEFAULT_FROM: &'static str = "2001:db8:100::/64";
     const PROVIDER_WAN_ADDR: &'static str = "2001:db8:ffff:1::2/64";
     const UPSTREAM_WAN_ADDR: &'static str = "2001:db8:ffff:1::1/64";
+    const ON_LINK_PROVIDER_WAN_ADDR: &'static str = "2001:db8:100::2/64";
+    const ON_LINK_UPSTREAM_WAN_ADDR: &'static str = "2001:db8:100::1/64";
     const UPSTREAM_SERVER_ADDR: &'static str = "2001:db8:ffff:2::1/64";
     const SERVER_ADDR: &'static str = "2001:db8:ffff:2::100/64";
     const SERVER_IP: &'static str = "2001:db8:ffff:2::100";
 
-    fn setup() -> Self {
+    fn setup_with_topology(topology: PublicIpv6LabTopology) -> Self {
         prepare_linux_namespaces();
 
         del_netns(Self::UPSTREAM_NS);
@@ -544,13 +552,23 @@ impl PublicIpv6Lab {
             Self::SERVER_BRIDGE,
         );
 
+        let (provider_wan_addr, upstream_wan_addr) = match topology {
+            PublicIpv6LabTopology::DelegatedPrefix => {
+                (Self::PROVIDER_WAN_ADDR, Self::UPSTREAM_WAN_ADDR)
+            }
+            PublicIpv6LabTopology::OnLinkPrefix => (
+                Self::ON_LINK_PROVIDER_WAN_ADDR,
+                Self::ON_LINK_UPSTREAM_WAN_ADDR,
+            ),
+        };
+
         run_ip_in_ns(
             Self::PROVIDER_NS,
-            &["addr", "add", Self::PROVIDER_WAN_ADDR, "dev", "pubwan0"],
+            &["addr", "add", provider_wan_addr, "dev", "pubwan0"],
         );
         run_ip_in_ns(
             Self::UPSTREAM_NS,
-            &["addr", "add", Self::UPSTREAM_WAN_ADDR, "dev", "upwan0"],
+            &["addr", "add", upstream_wan_addr, "dev", "upwan0"],
         );
         run_ip_in_ns(
             Self::UPSTREAM_NS,
@@ -561,37 +579,56 @@ impl PublicIpv6Lab {
             &["addr", "add", Self::SERVER_ADDR, "dev", "srv0"],
         );
 
-        run_ip_in_ns(
-            Self::PROVIDER_NS,
-            &["link", "add", "pubprefix0", "type", "dummy"],
-        );
-        run_ip_in_ns(Self::PROVIDER_NS, &["link", "set", "pubprefix0", "up"]);
-        run_ip_in_ns(
-            Self::PROVIDER_NS,
-            &[
-                "-6",
-                "route",
-                "add",
-                Self::PROVIDER_PREFIX,
-                "dev",
-                "pubprefix0",
-            ],
-        );
-        run_ip_in_ns(
-            Self::PROVIDER_NS,
-            &[
-                "-6",
-                "route",
-                "add",
-                "default",
-                "from",
-                Self::PROVIDER_DEFAULT_FROM,
-                "via",
-                "2001:db8:ffff:1::1",
-                "dev",
-                "pubwan0",
-            ],
-        );
+        match topology {
+            PublicIpv6LabTopology::DelegatedPrefix => {
+                run_ip_in_ns(
+                    Self::PROVIDER_NS,
+                    &["link", "add", "pubprefix0", "type", "dummy"],
+                );
+                run_ip_in_ns(Self::PROVIDER_NS, &["link", "set", "pubprefix0", "up"]);
+                run_ip_in_ns(
+                    Self::PROVIDER_NS,
+                    &[
+                        "-6",
+                        "route",
+                        "add",
+                        Self::PROVIDER_PREFIX,
+                        "dev",
+                        "pubprefix0",
+                    ],
+                );
+                run_ip_in_ns(
+                    Self::PROVIDER_NS,
+                    &[
+                        "-6",
+                        "route",
+                        "add",
+                        "default",
+                        "from",
+                        Self::PROVIDER_DEFAULT_FROM,
+                        "via",
+                        "2001:db8:ffff:1::1",
+                        "dev",
+                        "pubwan0",
+                    ],
+                );
+            }
+            PublicIpv6LabTopology::OnLinkPrefix => {
+                run_ip_in_ns(
+                    Self::PROVIDER_NS,
+                    &[
+                        "-6",
+                        "route",
+                        "add",
+                        "default",
+                        "via",
+                        "2001:db8:100::1",
+                        "dev",
+                        "pubwan0",
+                    ],
+                );
+            }
+        }
 
         run_ip_in_ns(
             Self::SERVER_NS,
@@ -606,19 +643,21 @@ impl PublicIpv6Lab {
                 "srv0",
             ],
         );
-        run_ip_in_ns(
-            Self::UPSTREAM_NS,
-            &[
-                "-6",
-                "route",
-                "add",
-                Self::PROVIDER_PREFIX,
-                "via",
-                "2001:db8:ffff:1::2",
-                "dev",
-                "upwan0",
-            ],
-        );
+        if matches!(topology, PublicIpv6LabTopology::DelegatedPrefix) {
+            run_ip_in_ns(
+                Self::UPSTREAM_NS,
+                &[
+                    "-6",
+                    "route",
+                    "add",
+                    Self::PROVIDER_PREFIX,
+                    "via",
+                    "2001:db8:ffff:1::2",
+                    "dev",
+                    "upwan0",
+                ],
+            );
+        }
 
         run_sysctl_in_ns(Self::PROVIDER_NS, "net.ipv6.conf.all.forwarding=1");
         run_sysctl_in_ns(Self::UPSTREAM_NS, "net.ipv6.conf.all.forwarding=1");
@@ -672,7 +711,15 @@ fn get_public_ipv6_config(
 async fn init_public_ipv6_two_node(
     client_inst_id: uuid::Uuid,
 ) -> (PublicIpv6Lab, Instance, Instance) {
-    let lab = PublicIpv6Lab::setup();
+    init_public_ipv6_two_node_with_topology(client_inst_id, PublicIpv6LabTopology::DelegatedPrefix)
+        .await
+}
+
+async fn init_public_ipv6_two_node_with_topology(
+    client_inst_id: uuid::Uuid,
+    topology: PublicIpv6LabTopology,
+) -> (PublicIpv6Lab, Instance, Instance) {
+    let lab = PublicIpv6Lab::setup_with_topology(topology);
 
     let provider_cfg = get_public_ipv6_config(
         "provider_public_ipv6",
@@ -756,6 +803,13 @@ fn addr_exists_in_ns(ns: &str, dev: &str, needle: &str) -> bool {
     run_ip_in_ns_output(ns, &["-6", "addr", "show", "dev", dev]).contains(needle)
 }
 
+fn ndp_proxy_exists_in_ns(ns: &str, dev: &str, addr: std::net::Ipv6Addr) -> bool {
+    let addr = addr.to_string();
+    run_ip_in_ns_output(ns, &["-6", "neigh", "show", "proxy", "dev", dev])
+        .lines()
+        .any(|line| line.split_whitespace().next() == Some(addr.as_str()))
+}
+
 #[tokio::test]
 #[serial_test::serial]
 pub async fn public_ipv6_auto_addr_end_to_end() {
@@ -878,6 +932,67 @@ pub async fn public_ipv6_auto_addr_end_to_end() {
     drop_insts(vec![provider, client]).await;
 }
 
+#[tokio::test]
+#[serial_test::serial]
+pub async fn public_ipv6_auto_addr_on_link_ndp_proxy_end_to_end() {
+    let client_id = uuid::Uuid::parse_str("44444444-4444-4444-4444-444444444444").unwrap();
+    let (_lab, provider, client) =
+        init_public_ipv6_two_node_with_topology(client_id, PublicIpv6LabTopology::OnLinkPrefix)
+            .await;
+
+    wait_for_condition(
+        || async {
+            provider
+                .get_global_ctx()
+                .get_advertised_ipv6_public_addr_prefix()
+                == Some(PublicIpv6Lab::PROVIDER_PREFIX.parse().unwrap())
+        },
+        Duration::from_secs(10),
+    )
+    .await;
+
+    let leased = wait_for_public_ipv6_addr(&client).await;
+    wait_for_public_ipv6_route(&provider, leased).await;
+
+    wait_for_condition(
+        || async {
+            addr_exists_in_ns(
+                PublicIpv6Lab::CLIENT_NS,
+                PublicIpv6Lab::CLIENT_TUN,
+                &leased.to_string(),
+            ) && route_exists_in_ns(
+                PublicIpv6Lab::PROVIDER_NS,
+                &format!("{} dev {}", leased.address(), PublicIpv6Lab::PROVIDER_TUN),
+            )
+        },
+        Duration::from_secs(10),
+    )
+    .await;
+
+    wait_for_condition(
+        || async {
+            ndp_proxy_exists_in_ns(PublicIpv6Lab::PROVIDER_NS, "pubwan0", leased.address())
+        },
+        Duration::from_secs(20),
+    )
+    .await;
+
+    wait_for_condition(
+        || async {
+            ping6_test(
+                PublicIpv6Lab::SERVER_NS,
+                leased.address().to_string().as_str(),
+                None,
+            )
+            .await
+        },
+        Duration::from_secs(20),
+    )
+    .await;
+
+    drop_insts(vec![provider, client]).await;
+}
+
 #[tokio::test]
 #[serial_test::serial]
 pub async fn public_ipv6_auto_addr_reconnect_reuses_same_address() {

easytier/src/tunnel/mpsc.rs

@@ -1,13 +1,6 @@
 // this mod wrap tunnel to a mpsc tunnel, based on crossbeam_channel
 
-use std::{
-    cell::UnsafeCell,
-    pin::Pin,
-    sync::Arc,
-    sync::atomic::{AtomicBool, Ordering},
-    task::Poll,
-    time::Duration,
-};
+use std::{pin::Pin, time::Duration};
 
 use anyhow::Context;
 use tokio::time::timeout;
@@ -18,123 +11,34 @@ use super::{Tunnel, TunnelError, ZCPacketSink, ZCPacketStream, packet_def::ZCPac
 
 use tokio::sync::mpsc::{Receiver, Sender, channel, error::TrySendError};
 use tokio_util::task::AbortOnDropHandle;
+// use tachyonix::{channel, Receiver, Sender, TrySendError};
 
 use futures::SinkExt;
 
-/// A simple spinlock protecting a sink. The guard is Send because it only
-/// contains an atomic flag reference (no lifetime-tied borrow like MutexGuard).
-struct SpinSink {
-    locked: AtomicBool,
-    sink: UnsafeCell<Pin<Box<dyn ZCPacketSink>>>,
-}
-
-// SAFETY: access is serialized by the spinlock.
-unsafe impl Send for SpinSink {}
-unsafe impl Sync for SpinSink {}
-
-struct SpinGuard<'a> {
-    spin: &'a SpinSink,
-}
-
-impl<'a> SpinGuard<'a> {
-    fn as_mut(&mut self) -> Pin<&mut dyn ZCPacketSink> {
-        // SAFETY: we hold the spinlock, so we have exclusive access
-        let sink = unsafe { &mut *self.spin.sink.get() };
-        sink.as_mut()
-    }
-}
-
-impl Drop for SpinGuard<'_> {
-    fn drop(&mut self) {
-        self.spin.locked.store(false, Ordering::Release);
-    }
-}
-
-impl SpinSink {
-    fn new(sink: Pin<Box<dyn ZCPacketSink>>) -> Self {
-        Self {
-            locked: AtomicBool::new(false),
-            sink: UnsafeCell::new(sink),
-        }
-    }
-
-    fn try_lock(&self) -> Option<SpinGuard<'_>> {
-        if self
-            .locked
-            .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
-            .is_ok()
-        {
-            Some(SpinGuard { spin: self })
-        } else {
-            None
-        }
-    }
-}
-
 #[derive(Clone)]
-pub struct MpscTunnelSender {
-    channel_tx: Option<Sender<ZCPacket>>,
-    direct_sink: Option<Arc<SpinSink>>,
-}
+pub struct MpscTunnelSender(Sender<ZCPacket>);
 
 impl MpscTunnelSender {
     pub async fn send(&self, item: ZCPacket) -> Result<(), TunnelError> {
-        if let Some(sink) = &self.direct_sink {
-            // Sync fast path: no await needed, returns immediately
-            if let Some(mut guard) = sink.try_lock() {
-                let waker = futures::task::noop_waker();
-                let mut cx = std::task::Context::from_waker(&waker);
-                match guard.as_mut().poll_ready(&mut cx) {
-                    Poll::Ready(Ok(())) => {
-                        guard.as_mut().start_send(item)?;
-                        // poll_flush may return Pending when the consumer task hasn't
-                        // drained the ring yet. The data is already in the ring buffer
-                        // and will be consumed — treat Pending as success.
-                        match guard.as_mut().poll_flush(&mut cx) {
-                            Poll::Ready(Err(e)) => return Err(e),
-                            _ => return Ok(()),
-                        }
-                    }
-                    Poll::Ready(Err(e)) => return Err(e),
-                    Poll::Pending => return Err(TunnelError::BufferFull),
-                }
-            }
-            return Err(TunnelError::BufferFull);
-        }
-
-        // Channel mode: async with backpressure
-        self.send_async(item).await
+        self.0.send(item).await.with_context(|| "send error")?;
+        Ok(())
     }
 
     pub fn try_send(&self, item: ZCPacket) -> Result<(), TunnelError> {
-        let tx = self.channel_tx.as_ref().ok_or(TunnelError::Shutdown)?;
-        tx.try_send(item).map_err(|e| match e {
+        self.0.try_send(item).map_err(|e| match e {
             TrySendError::Full(_) => TunnelError::BufferFull,
             TrySendError::Closed(_) => TunnelError::Shutdown,
         })
     }
-
-    pub async fn send_async(&self, item: ZCPacket) -> Result<(), TunnelError> {
-        let tx = self.channel_tx.as_ref().ok_or(TunnelError::Shutdown)?;
-        match tx.try_send(item) {
-            Ok(()) => Ok(()),
-            Err(TrySendError::Full(item)) => {
-                tx.send(item).await.with_context(|| "send error")?;
-                Ok(())
-            }
-            Err(TrySendError::Closed(_)) => Err(TunnelError::Shutdown),
-        }
-    }
 }
 
 pub struct MpscTunnel<T> {
     tx: Option<Sender<ZCPacket>>,
-    direct_sink: Option<Arc<SpinSink>>,
 
     tunnel: T,
     stream: Option<Pin<Box<dyn ZCPacketStream>>>,
 
-    task: Option<AbortOnDropHandle<()>>,
+    task: AbortOnDropHandle<()>,
 }
 
 impl<T: Tunnel> MpscTunnel<T> {
@@ -156,21 +60,9 @@ impl<T: Tunnel> MpscTunnel<T> {
 
         Self {
             tx: Some(tx),
-            direct_sink: None,
             tunnel,
             stream: Some(stream),
-            task: Some(AbortOnDropHandle::new(task)),
-        }
-    }
-
-    pub fn new_direct(tunnel: T) -> Self {
-        let (stream, sink) = tunnel.split();
-        Self {
-            tx: None,
-            direct_sink: Some(Arc::new(SpinSink::new(sink))),
-            tunnel,
-            stream: Some(stream),
-            task: None,
+            task: AbortOnDropHandle::new(task),
         }
     }
 
@@ -232,18 +124,12 @@ impl<T: Tunnel> MpscTunnel<T> {
     }
 
     pub fn get_sink(&self) -> MpscTunnelSender {
-        MpscTunnelSender {
-            channel_tx: self.tx.as_ref().cloned(),
-            direct_sink: self.direct_sink.clone(),
-        }
+        MpscTunnelSender(self.tx.as_ref().unwrap().clone())
     }
 
     pub fn close(&mut self) {
         self.tx.take();
-        self.direct_sink.take();
-        if let Some(task) = self.task.take() {
-            task.abort();
-        }
+        self.task.abort();
     }
 
     pub fn tunnel_info(&self) -> Option<TunnelInfo> {
@@ -303,7 +189,7 @@ mod tests {
             for i in 0..1000000 {
                 tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
                 let a = sink1
-                    .send_async(ZCPacket::new_with_payload("hello".as_bytes()))
+                    .send(ZCPacket::new_with_payload("hello".as_bytes()))
                     .await;
                 if a.is_err() {
                     tracing::info!(?a, "t2 exit with err");
@@ -323,7 +209,7 @@ mod tests {
             for i in 0..1000000 {
                 tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
                 let a = sink2
-                    .send_async(ZCPacket::new_with_payload("hello2".as_bytes()))
+                    .send(ZCPacket::new_with_payload("hello2".as_bytes()))
                     .await;
                 if a.is_err() {
                     tracing::info!(?a, "t3 exit with err");

easytier/src/tunnel/packet_def.rs

@@ -1,3 +1,4 @@
+use bytes::Buf;
 use bytes::Bytes;
 use bytes::BytesMut;
 use zerocopy::AsBytes;
@@ -486,7 +487,7 @@ impl ZCPacket {
         let total_len = payload_off + payload.len();
         ret.inner.reserve(total_len);
         unsafe { ret.inner.set_len(total_len) };
-        ret.mut_payload()[..payload.len()].copy_from_slice(payload);
+        ret.mut_payload().copy_from_slice(payload);
         ret
     }
 
@@ -587,7 +588,8 @@ impl ZCPacket {
     }
 
     pub fn payload_bytes(mut self) -> BytesMut {
-        self.inner.split_off(self.payload_offset())
+        self.inner.advance(self.payload_offset());
+        self.inner
     }
 
     pub fn peer_manager_header(&self) -> Option<&PeerManagerHeader> {
@@ -652,11 +654,12 @@ impl ZCPacket {
     }
 
     pub fn tunnel_payload_bytes(mut self) -> BytesMut {
-        self.inner.split_off(
+        self.inner.advance(
             self.packet_type
                 .get_packet_offsets()
                 .peer_manager_header_offset,
-        )
+        );
+        self.inner
     }
 
     pub fn convert_type(mut self, target_packet_type: ZCPacketType) -> Self {
@@ -702,7 +705,8 @@ impl ZCPacket {
             return Self::new_from_buf(buf, target_packet_type);
         }
 
-        Self::new_from_buf(self.inner.split_off(new_offset), target_packet_type)
+        self.inner.advance(new_offset);
+        Self::new_from_buf(self.inner, target_packet_type)
     }
 
     pub fn into_bytes(self) -> Bytes {
@@ -748,8 +752,10 @@ impl ZCPacket {
         let foreign_hdr_len = hdr.get_header_len();
 
         Self::new_from_buf(
-            self.inner
-                .split_off(foreign_hdr_len + self.payload_offset()),
+            {
+                self.inner.advance(foreign_hdr_len + self.payload_offset());
+                self.inner
+            },
             ZCPacketType::DummyTunnel,
         )
     }