test: validate the explicit test workflow artifact

Repeat the failing Ethernet send test on an Ubuntu hosted runner.
Record link events, packets, and post-send drop counters without
changing the original test binary or introducing a pre-send delay.
Continue after MAC assertion failures to capture the original timeout.

.github/workflows/bpf-diagnostic.yml

@@ -0,0 +1,216 @@
+name: Validate BPF veth initialization
+
+on:
+  push:
+    branches: ["diagnose/bpf-2710"]
+
+permissions:
+  contents: read
+  actions: read
+
+jobs:
+  diagnose:
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    steps:
+      - uses: actions/checkout@v5
+      - name: Wait for the fixed test artifact
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          for attempt in $(seq 1 180); do
+            if gh api repos/EasyTier/EasyTier/actions/runs/38869099758/artifacts --jq '.artifacts[].name' | grep -qx tests; then
+              exit 0
+            fi
+            sleep 5
+          done
+          exit 1
+      - uses: actions/download-artifact@v4
+        with:
+          name: tests
+          run-id: 38869099758
+          github-token: ${{ github.token }}
+      - name: Prepare fixed CI binary
+        run: |
+          mkdir -p diagnostics
+          tar --zstd -xf tests.tar.zst
+          python3 - <<'PY'
+          import pathlib, re
+          binaries = [p for p in pathlib.Path('target/debug/deps').iterdir() if re.fullmatch(r'easytier-[0-9a-f]+', p.name)]
+          assert len(binaries) == 1, binaries
+          pathlib.Path('diagnostics/binary-path').write_text(str(binaries[0].resolve()))
+          PY
+          sudo apt-get update -qq
+          sudo apt-get install -y -qq strace tcpdump
+          uname -a > diagnostics/environment.txt
+          systemctl is-active systemd-udevd systemd-networkd NetworkManager >> diagnostics/environment.txt || true
+          sudo ip -ts -s -d monitor link > diagnostics/links.log &
+          sudo tcpdump -i any -U -w diagnostics/frames.pcap 'tcp port 23456' > diagnostics/tcpdump.log 2>&1 &
+          sudo bash -c 'echo "p:bpf_noop noop_enqueue" >> /sys/kernel/tracing/kprobe_events; echo 1 > /sys/kernel/tracing/events/kprobes/bpf_noop/enable' || true
+          sudo cat /sys/kernel/tracing/trace_pipe > diagnostics/kernel-trace.log &
+      - name: Build send observation hook
+        run: |
+          cat > diagnostics/send-observer.c <<'EOF'
+          #define _GNU_SOURCE
+          #include <dlfcn.h>
+          #include <stdio.h>
+          #include <sys/socket.h>
+          #include <netpacket/packet.h>
+          #include <net/if.h>
+          #include <errno.h>
+          #include <time.h>
+          static ssize_t (*real_sendto)(int,const void*,size_t,int,const struct sockaddr*,socklen_t);
+          __attribute__((constructor)) static void initialize(void) {
+              real_sendto = dlsym(RTLD_NEXT, "sendto");
+          }
+          static void read_value(const char *name, const char *key, char *out) {
+              char path[256];
+              snprintf(path, sizeof(path), "/sys/class/net/%s/%s", name, key);
+              FILE *f = fopen(path, "r");
+              if (f) { if (!fgets(out, 64, f)) out[0] = 0; fclose(f); }
+          }
+          ssize_t sendto(int fd, const void *buf, size_t len, int flags,
+                         const struct sockaddr *addr, socklen_t addrlen) {
+              ssize_t ret = real_sendto(fd, buf, len, flags, addr, addrlen);
+              int saved_errno = errno;
+              if (addr && addr->sa_family == AF_PACKET) {
+                  const struct sockaddr_ll *ll = (const struct sockaddr_ll *)addr;
+                  char name[IF_NAMESIZE] = {0}, drops[64] = {0}, state[64] = {0}, carrier[64] = {0};
+                  char packets[64] = {0};
+                  struct timespec timestamp;
+                  clock_gettime(CLOCK_REALTIME, &timestamp);
+                  if_indextoname(ll->sll_ifindex, name);
+                  read_value(name, "statistics/tx_dropped", drops);
+                  read_value(name, "operstate", state);
+                  read_value(name, "carrier", carrier);
+                  read_value(name, "statistics/tx_packets", packets);
+                  fprintf(stderr, "SEND_OBSERVE time=%ld.%09ld dev=%s len=%zu ret=%zd errno=%d tx_dropped=%s operstate=%s carrier=%s tx_packets=%s\n", timestamp.tv_sec, timestamp.tv_nsec, name, len, ret, saved_errno, drops, state, carrier, packets);
+              }
+              errno = saved_errno;
+              return ret;
+          }
+          EOF
+          gcc -shared -fPIC -O2 diagnostics/send-observer.c -ldl -o diagnostics/send-observer.so
+      - name: Repeat fixed test with post-send observation
+        run: |
+          sudo -E python3 -u - <<'PY'
+          import os, pathlib, subprocess, time
+          root = pathlib.Path.cwd()
+          env = os.environ.copy()
+          env['EASYTIER_LINUX_BPF_INTEGRATION'] = 'required'
+          env['LD_PRELOAD'] = str(root / 'diagnostics/send-observer.so')
+          binary = (root / 'diagnostics/binary-path').read_text()
+          case = 'socket::fake_tcp::netfilter::linux_bpf::tests::linux_bpf_tun_sends_ipv4_frame_to_ethernet_interface'
+          start = time.monotonic()
+          failures = 0
+          timeouts = 0
+          for i in range(1, 21):
+              result = subprocess.run([binary, case, '--exact', '--nocapture'], env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
+              if i <= 3 or result.returncode:
+                  (root / f'diagnostics/attempt-{i}.log').write_text(result.stdout)
+              if result.returncode:
+                  failures += 1
+                  print('REPRODUCED', i, 'elapsed', time.monotonic() - start, flush=True)
+                  print(result.stdout, flush=True)
+                  if 'timed out waiting for Ethernet frame' in result.stdout:
+                      timeouts += 1
+                  elif 'assertion `left == right` failed' not in result.stdout:
+                      break
+                  if timeouts == 3:
+                      break
+              if i % 100 == 0:
+                  print('PASS', i, 'elapsed', time.monotonic() - start, flush=True)
+          else:
+              print('COMPLETED 20 ATTEMPTS', flush=True)
+          print('TOTAL FAILURES', failures, 'TIMEOUTS', timeouts, flush=True)
+          assert failures == 0
+          PY
+      - name: Repeat fixed binary without observation hooks
+        run: |
+          sudo -E python3 -u - <<'PY'
+          import os, pathlib, subprocess, time
+          root = pathlib.Path.cwd()
+          env = os.environ.copy()
+          env['EASYTIER_LINUX_BPF_INTEGRATION'] = 'required'
+          binary = (root / 'diagnostics/binary-path').read_text()
+          case = 'socket::fake_tcp::netfilter::linux_bpf::tests::linux_bpf_tun_sends_ipv4_frame_to_ethernet_interface'
+          start = time.monotonic()
+          failures = 0
+          for i in range(1, 3001):
+              result = subprocess.run([binary, case, '--exact', '--nocapture'], env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
+              if result.returncode:
+                  failures += 1
+                  (root / f'diagnostics/baseline-{i}.log').write_text(result.stdout)
+                  if 'timed out waiting for Ethernet frame' in result.stdout:
+                      print('ORIGINAL TIMEOUT', i, 'elapsed', time.monotonic() - start, flush=True)
+                      print(result.stdout, flush=True)
+                      break
+                  if 'assertion `left == right` failed' not in result.stdout:
+                      print('UNEXPECTED', i, result.stdout, flush=True)
+                      break
+              if i % 500 == 0:
+                  print('BASELINE', i, 'failures', failures, 'elapsed', time.monotonic() - start, flush=True)
+          else:
+              print('COMPLETED 3000 FIXED ATTEMPTS', flush=True)
+          assert failures == 0
+          PY
+      - name: Validate fixed fixture under worker scheduling pressure
+        run: |
+          cat > diagnostics/cpu-pressure.c <<'EOF'
+          #define _GNU_SOURCE
+          #include <sched.h>
+          #include <stdlib.h>
+          #include <time.h>
+          #include <stdio.h>
+          int main(int argc, char **argv) {
+              cpu_set_t mask;
+              CPU_ZERO(&mask);
+              CPU_SET(atoi(argv[1]), &mask);
+              if (sched_setaffinity(0, sizeof(mask), &mask)) { perror("affinity"); return 1; }
+              struct sched_param param = {.sched_priority = 10};
+              if (sched_setscheduler(0, SCHED_FIFO, &param)) { perror("scheduler"); return 1; }
+              struct timespec start, now;
+              clock_gettime(CLOCK_MONOTONIC, &start);
+              do { clock_gettime(CLOCK_MONOTONIC, &now); } while (now.tv_sec - start.tv_sec < 300);
+              return 0;
+          }
+          EOF
+          gcc -O2 diagnostics/cpu-pressure.c -o diagnostics/cpu-pressure
+          sudo -E python3 -u - <<'PY'
+          import os, pathlib, subprocess, time
+          root = pathlib.Path.cwd()
+          env = os.environ.copy()
+          env['EASYTIER_LINUX_BPF_INTEGRATION'] = 'required'
+          env['LD_PRELOAD'] = str(root / 'diagnostics/send-observer.so')
+          binary = (root / 'diagnostics/binary-path').read_text()
+          case = 'socket::fake_tcp::netfilter::linux_bpf::tests::linux_bpf_tun_sends_ipv4_frame_to_ethernet_interface'
+          hogs = []
+          try:
+              os.sched_setscheduler(0, os.SCHED_FIFO, os.sched_param(50))
+              for cpu in sorted(os.sched_getaffinity(0)):
+                  hogs.append(subprocess.Popen([str(root / 'diagnostics/cpu-pressure'), str(cpu)]))
+              failures = 0
+              timeouts = 0
+              for i in range(1, 51):
+                  assert all(proc.poll() is None for proc in hogs), 'CPU load ended early'
+                  result = subprocess.run(['chrt', '-f', '20', binary, case, '--exact', '--nocapture'], env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=10)
+                  (root / f'diagnostics/pressure-fixed-{i}.log').write_text(result.stdout)
+                  failures += bool(result.returncode)
+                  timeouts += 'timed out waiting for Ethernet frame' in result.stdout
+                  print('PRESSURE fixed', i, result.returncode, result.stdout, flush=True)
+                  assert 'tx_dropped=0\n' in result.stdout
+              assert all(proc.poll() is None for proc in hogs), 'CPU load ended early'
+              print('PRESSURE TOTAL fixed', 'failures', failures, 'timeouts', timeouts, flush=True)
+              assert failures == 0
+          finally:
+              for proc in hogs:
+                  proc.kill()
+              for proc in hogs:
+                  proc.wait()
+              os.sched_setscheduler(0, os.SCHED_OTHER, os.sched_param(0))
+          PY
+      - uses: actions/upload-artifact@v4
+        if: always()
+        with:
+          name: bpf-diagnostics
+          path: diagnostics

.github/workflows/ohos.yml

@@ -138,8 +138,7 @@ jobs:
           set -euo pipefail
           cargo_version=$(cargo metadata --format-version 1 --no-deps \
             --manifest-path easytier/Cargo.toml | jq -r '.packages[0].version')
-          # Rolling tags such as nightly are not package versions.
-          last_tag=$(git describe --tags --match 'v[0-9]*' --abbrev=0 HEAD 2>/dev/null || true)
+          last_tag=$(git describe --tags --abbrev=0 HEAD 2>/dev/null || true)
           if [ -n "$last_tag" ]; then
             base_version=$(printf '%s\n' "$cargo_version" "${last_tag#v}" \
               | sort -V | tail -n 1)

docs/core-architecture.md

@@ -552,13 +552,6 @@ optional `tracing` feature selects the tokio-console subscriber integration;
 only that diagnostic profile pulls the main crate's `tracing-subscriber` and
 `console-subscriber` dependencies.
 
-The native CLI enables the `CORE` and `easytier_core::instance::manager` console
-targets by default. Instance startup failures are reported immediately with the
-instance ID and complete error chain, including the listener URL for bind
-failures. When all instances stop, the CLI's failure summary retains each failed
-instance's ID and saved error. Explicit console levels and `RUST_LOG` directives
-still control which records are emitted.
-
 Contrib applications and platform integrations may have independent logging
 requirements and are not implicitly wired to the native process logger.
 

easytier-core/src/instance/manager.rs

@@ -387,11 +387,7 @@ where
         let instance_state_changes = self.instance_state_changes.clone();
         runtime.spawn(async move {
             if let Err(error) = instance.start().await {
-                tracing::error!(
-                    error = %format_args!("{error:#}"),
-                    %instance_id,
-                    "instance failed to start"
-                );
+                tracing::error!(%error, %instance_id, "instance failed to start");
             }
             instance_state_changes.mark_changed();
             notifier.notify_one();

easytier-core/src/listener/mod.rs

@@ -342,11 +342,10 @@ where
     Accepted: Send + 'static,
 {
     let mut listener = creator();
-    listener
-        .listen()
-        .await
-        .with_context(|| format!("failed to listen on {}", listener.local_url()))?;
-    Ok(listener)
+    match listener.listen().await {
+        Ok(()) => Ok(listener),
+        Err(error) => Err(error),
+    }
 }
 
 async fn run_listener<Accepted, H>(
@@ -783,8 +782,7 @@ mod tests {
             true,
         );
 
-        let error = manager.run().await.unwrap_err();
-        assert!(format!("{error:#}").contains("mock://second"));
+        assert!(manager.run().await.is_err());
         crate::foundation::time::sleep(Duration::from_millis(20)).await;
         assert!(handler.accepted.lock().unwrap().is_empty());
         assert!(

easytier-core/src/peers/attached.rs

@@ -199,7 +199,7 @@ impl AttachedPeerRuntime {
         let public_ipv6_runtime =
             CorePublicIpv6Runtime::new(runtime_config.clone(), Arc::new(()), Arc::new(()));
         let flags = peer_snapshot.flags.clone();
-        let mut peer_manager = PeerManagerCore::new(
+        let peer_manager = Arc::new(PeerManagerCore::new(
             PortablePeerManagerConfig {
                 snapshot: peer_snapshot,
                 route_algo: RouteAlgoType::Ospf,
@@ -214,9 +214,7 @@ impl AttachedPeerRuntime {
             Arc::new(()),
             None,
             Arc::new(()),
-        )?;
-        peer_manager.inherit_exit_nodes(&network_peer_manager);
-        let peer_manager = Arc::new(peer_manager);
+        )?);
         if let Err(error) = peer_manager
             .follow_network_policy(network_runtime_config, config.groups)
             .await
@@ -623,176 +621,6 @@ mod tests {
         peer_manager_with_acl(Vec::new())
     }
 
-    async fn connect_exit_peer(
-        network: &PeerManagerCore,
-        store: &CoreRuntimeConfigStore,
-        address: Ipv4Addr,
-    ) -> (Arc<PeerManagerCore>, HostPacketReceiver) {
-        let mut snapshot = store.snapshot().peer.as_ref().clone();
-        snapshot.runtime.core.routes.ipv4 = Some(IpPrefix {
-            address: address.into(),
-            prefix_len: 24,
-        });
-        if let Some(secure) = snapshot.runtime.secure_mode.as_mut() {
-            let private = StaticSecret::from([address.octets()[3]; 32]);
-            secure.local_private_key = Some(BASE64_STANDARD.encode(private.as_bytes()));
-            secure.local_public_key =
-                Some(BASE64_STANDARD.encode(PublicKey::from(&private).as_bytes()));
-        }
-        let flags = snapshot.flags.clone();
-        let store =
-            CoreRuntimeConfigStore::new(CoreRuntimeConfig::default(), Arc::new(snapshot.clone()));
-        let (sender, receiver) = host_packet_channel();
-        let public_ipv6 = CorePublicIpv6Runtime::new(store.clone(), Arc::new(()), Arc::new(()));
-        let peer = Arc::new(
-            PeerManagerCore::new(
-                PortablePeerManagerConfig {
-                    snapshot,
-                    route_algo: RouteAlgoType::Ospf,
-                    exit_nodes: Vec::new(),
-                    foreign_context_default_flags: flags,
-                },
-                Vec::new(),
-                store,
-                Arc::new(()),
-                sender,
-                public_ipv6,
-                Arc::new(()),
-                None,
-                Arc::new(()),
-            )
-            .unwrap(),
-        );
-        peer.run().await.unwrap();
-        let (server, client) = create_ring_tunnel_pair();
-        let (server_result, client_result) = tokio::join!(
-            network.add_tunnel_as_server(server),
-            peer.add_client_tunnel(client),
-        );
-        server_result.unwrap();
-        client_result.unwrap();
-        (peer, receiver)
-    }
-
-    async fn check_attached_peer_exit_nodes(secure: bool) {
-        let (network, store) = peer_manager_with_acl_and_secure(Vec::new(), secure);
-        network.run().await.unwrap();
-        let first_ip = Ipv4Addr::new(10, 82, 0, 10);
-        let second_ip = Ipv4Addr::new(10, 82, 0, 11);
-        let (first, mut first_packets) = connect_exit_peer(&network, &store, first_ip).await;
-        let (second, mut second_packets) = connect_exit_peer(&network, &store, second_ip).await;
-        network.update_exit_nodes(vec![first_ip.into()]).await;
-        let client_ip = Ipv4Addr::new(10, 82, 0, 2);
-        let attached = AttachedPeerRuntime::connect(
-            network.clone(),
-            store,
-            AttachedPeerConfig {
-                name: "exit-node-client".to_owned(),
-                virtual_ip: attached_ipv4(client_ip),
-                groups: vec!["ops".to_owned()],
-                identity_private_key: [2; 32],
-            },
-        )
-        .await
-        .unwrap();
-        for (exit, address) in [(&first, first_ip), (&second, second_ip)] {
-            wait_for_route(&attached.peer_manager, exit.my_peer_id(), address, true).await;
-            wait_for_route(exit, attached.peer_id(), client_ip, true).await;
-        }
-
-        let public_ip = Ipv4Addr::new(8, 8, 8, 8);
-        let mut request = vec![0u8; 28];
-        request[0] = 0x45;
-        request[2..4].copy_from_slice(&28u16.to_be_bytes());
-        request[8] = 64;
-        request[9] = 1;
-        request[12..16].copy_from_slice(&client_ip.octets());
-        request[16..20].copy_from_slice(&public_ip.octets());
-        request[20] = 8;
-
-        for (exit, address, packets) in [
-            (&first, first_ip, &mut first_packets),
-            (&second, second_ip, &mut second_packets),
-        ] {
-            if address == second_ip {
-                network.update_exit_nodes(vec![second_ip.into()]).await;
-            }
-            assert_eq!(
-                attached
-                    .peer_manager
-                    .get_msg_dst_peer(&public_ip.into())
-                    .await,
-                (vec![exit.my_peer_id()], true),
-            );
-            attached.send_packet(&request).await.unwrap();
-            let packet = tokio::time::timeout(Duration::from_secs(5), packets.recv())
-                .await
-                .expect("exit node did not receive the client packet")
-                .unwrap()
-                .into_core_packet();
-            assert_eq!(packet.payload(), request.as_slice());
-            let header = packet.peer_manager_header().unwrap();
-            assert!(header.is_exit_node());
-            assert_eq!(header.from_peer_id.get(), attached.peer_id());
-            assert_eq!(header.to_peer_id.get(), exit.my_peer_id());
-
-            let mut reply = request.clone();
-            reply[12..16].copy_from_slice(&public_ip.octets());
-            reply[16..20].copy_from_slice(&client_ip.octets());
-            reply[20] = 0;
-            exit.send_msg_by_ip(ZCPacket::new_with_payload(&reply), client_ip.into(), false)
-                .await
-                .unwrap();
-            let received = tokio::time::timeout(Duration::from_secs(5), attached.recv_packet())
-                .await
-                .expect("exit reply did not reach the client")
-                .unwrap();
-            assert_eq!(received.payload(), reply.as_slice());
-        }
-
-        network.update_exit_nodes(Vec::new()).await;
-        assert_eq!(
-            attached
-                .peer_manager
-                .get_msg_dst_peer(&public_ip.into())
-                .await,
-            (Vec::new(), false),
-        );
-        // Explicit mesh routes must still win, even with an exit configured.
-        network.update_exit_nodes(vec![second_ip.into()]).await;
-        assert_eq!(
-            attached
-                .peer_manager
-                .get_msg_dst_peer(&first_ip.into())
-                .await,
-            (vec![first.my_peer_id()], false),
-        );
-        request[16..20].copy_from_slice(&first_ip.octets());
-        attached.send_packet(&request).await.unwrap();
-        let packet = tokio::time::timeout(Duration::from_secs(5), first_packets.recv())
-            .await
-            .unwrap()
-            .unwrap()
-            .into_core_packet();
-        assert_eq!(packet.payload(), request.as_slice());
-        assert!(!packet.peer_manager_header().unwrap().is_exit_node());
-
-        attached.close().await;
-        first.clear_resources().await;
-        second.clear_resources().await;
-        network.clear_resources().await;
-    }
-
-    #[tokio::test]
-    async fn attached_peer_inherits_exit_nodes_and_updates() {
-        check_attached_peer_exit_nodes(false).await;
-    }
-
-    #[tokio::test]
-    async fn secure_attached_peer_inherits_exit_nodes_and_updates() {
-        check_attached_peer_exit_nodes(true).await;
-    }
-
     async fn wait_for_route(
         peer: &PeerManagerCore,
         attached_peer_id: PeerId,

easytier-core/src/peers/peer_manager.rs

@@ -1769,13 +1769,6 @@ impl PeerManagerCore {
         *self.exit_nodes.write().await = exit_nodes;
     }
 
-    /// Shares the network's current exit nodes, including later updates, while
-    /// keeping this peer's route lookup and packet identity independent.
-    pub(super) fn inherit_exit_nodes(&mut self, network: &Self) {
-        self.exit_nodes = network.exit_nodes.clone();
-        self.outbound_packet_router.exit_nodes = self.exit_nodes.clone();
-    }
-
     fn reload_acl(&self, acl: Option<&crate::proto::acl::Acl>) {
         self.acl_filter.reload_rules(acl);
     }

easytier-core/src/tunnel/secure_datagram.rs

@@ -1011,12 +1011,14 @@ mod tests {
         assert!(s.check_replay_for_test(1, 0, SecureDatagramDirection::AToB, now + 1));
 
         s.sync_root_key(root_key, 2, 2, true);
-        // Use the actual deadline so scheduling delays before sync do not affect the test.
-        let expires_at_ms = s.sync_rx_grace_expires_at_ms.load(Ordering::Relaxed);
-        assert!(s.check_replay_for_test(2, 0, SecureDatagramDirection::AToB, expires_at_ms - 1));
-        assert!(s.check_replay_for_test(0, 1, SecureDatagramDirection::AToB, expires_at_ms - 1));
+        assert!(s.check_replay_for_test(2, 0, SecureDatagramDirection::AToB, now + 2));
 
-        assert!(!s.check_replay_for_test(0, 2, SecureDatagramDirection::AToB, expires_at_ms));
+        assert!(!s.check_replay_for_test(
+            0,
+            1,
+            SecureDatagramDirection::AToB,
+            now + SecureDatagramSession::SYNC_RX_GRACE_AFTER_MS + 3
+        ));
     }
 
     #[test]

easytier-gui/src-tauri/tauri.windows.conf.json

@@ -7,12 +7,9 @@
             "./*.sys"
         ],
         "windows": {
-            "nsis": {
-                "installerHooks": "./windows/installer-hooks.nsh"
-            },
             "webviewInstallMode": {
                 "type": "embedBootstrapper"
             }
         }
     }
-}
+}

easytier-gui/src-tauri/windows/README.md

@@ -1,28 +0,0 @@
-# Windows installer process handling
-
-NSIS remains a current-user installer. Its helper requests elevation only
-when a running service or GUI requires it, and restores service startup and
-running state after installation. During an update it temporarily disables
-service startup because released EasyTier versions can trigger SCM recovery
-even after an explicit stop.
-
-Run `test-installer-processes.ps1` with Windows PowerShell as administrator
-on a test machine without an existing `easytier-gui` service. The test uses
-isolated executables and a service that reproduces the old exit behavior
-with failure recovery enabled. Run it with `-GuiOnly` from an unelevated
-shell to check the path that needs no elevation.
-
-## Known limitations
-
-If preparation on the welcome page has already started a helper, it is
-reused at installation time. Changing to another directory containing a
-running GUI, or relaunching the GUI between those pages, may still cause
-Tauri's default process check to fail. Rechecking those processes is a
-separate follow-up; it was recorded as a minor finding in the final review.
-
-Executable matching does not expand Windows 8.3 short-path aliases. An
-installation directory expressed using a short alias may not match the
-process image's full path. This was reproduced with LocalService's short
-temporary-directory path; the unprivileged GUI test passes when its
-temporary directory uses the full path. This is recorded as a minor
-follow-up. Use a full temporary-directory path when running the fixture.

easytier-gui/src-tauri/windows/installer-hooks.nsh

@@ -1,95 +0,0 @@
-!define EASYTIER_INSTALLER_SCRIPT "${__FILEDIR__}\installer-processes.ps1"
-
-; Tauri uninstalls the old version on its reinstall page, before PREINSTALL.
-; MUI consumes this callback on the first page (Welcome). Opening the installer
-; alone does not stop anything; the user must first leave the welcome page.
-!define MUI_PAGE_CUSTOMFUNCTION_LEAVE EasyTierPrepareInstall
-!define MUI_CUSTOMFUNCTION_GUIINIT EasyTierInstallerGuiInit
-
-Var EasyTierHelper
-Var EasyTierInstallerPid
-Var EasyTierDone
-Var EasyTierFailed
-
-!macro EasyTierPrepare RESTORE
-  ${If} $EasyTierHelper != ""
-    Return
-  ${EndIf}
-  InitPluginsDir
-  File "/oname=$PLUGINSDIR\installer-processes.ps1" "${EASYTIER_INSTALLER_SCRIPT}"
-  System::Call 'kernel32::GetCurrentProcessId() i .r0'
-  StrCpy $EasyTierInstallerPid $0
-  nsExec::ExecToStack '"$SYSDIR\WindowsPowerShell\v1.0\powershell.exe" -NoLogo -NoProfile -NonInteractive -ExecutionPolicy Bypass -File "$PLUGINSDIR\installer-processes.ps1" -InstallDir "$INSTDIR" -InstallerPid $EasyTierInstallerPid -RestoreService ${RESTORE}'
-  Pop $0
-  Pop $1
-  ${If} $0 != 0
-    MessageBox MB_OK|MB_ICONSTOP "Could not close EasyTier:$\r$\n$1" /SD IDOK
-    Abort
-  ${EndIf}
-  IntOp $1 $1 + 0
-  ${If} $1 != 0
-    System::Call 'kernel32::OpenEventW(i 0x100000, i 0, w "Local\EasyTierInstaller-$EasyTierInstallerPid-done") p .r0'
-    StrCpy $EasyTierDone $0
-    System::Call 'kernel32::OpenEventW(i 0x100000, i 0, w "Local\EasyTierInstaller-$EasyTierInstallerPid-failed") p .r0'
-    StrCpy $EasyTierFailed $0
-    ${If} $EasyTierDone == 0
-    ${OrIf} $EasyTierFailed == 0
-      MessageBox MB_OK|MB_ICONSTOP "Could not wait for EasyTier to finish updating." /SD IDOK
-      Abort
-    ${EndIf}
-    StrCpy $EasyTierHelper 1
-  ${EndIf}
-!macroend
-
-!macro EasyTierFinish
-  ${If} $EasyTierHelper != ""
-    System::Call 'kernel32::OpenEventW(i 2, i 0, w "Local\EasyTierInstaller-$EasyTierInstallerPid-finish") p .r0'
-    ${If} $0 != 0
-      System::Call 'kernel32::SetEvent(p r0)'
-      System::Call 'kernel32::CloseHandle(p r0)'
-    ${EndIf}
-    System::Call 'kernel32::WaitForSingleObject(p $EasyTierDone, i 65000) i .r0'
-    System::Call 'kernel32::WaitForSingleObject(p $EasyTierFailed, i 0) i .r1'
-    System::Call 'kernel32::CloseHandle(p $EasyTierDone)'
-    System::Call 'kernel32::CloseHandle(p $EasyTierFailed)'
-    StrCpy $EasyTierHelper ""
-    ${If} $0 != 0
-    ${OrIf} $1 != 258
-      MessageBox MB_OK|MB_ICONSTOP "Could not restore the EasyTier service. Please check its status in Windows Services." /SD IDOK
-      Abort
-    ${EndIf}
-  ${EndIf}
-!macroend
-
-Function EasyTierPrepareInstall
-  !insertmacro EasyTierPrepare 1
-FunctionEnd
-
-Function EasyTierInstallerGuiInit
-  ; Passive installs skip Welcome, but still run the old uninstaller page.
-  ${GetOptions} $CMDLINE "/P" $0
-  ${IfNot} ${Errors}
-    Call EasyTierPrepareInstall
-  ${EndIf}
-FunctionEnd
-
-Function un.EasyTierPrepareUninstall
-  !insertmacro EasyTierPrepare 0
-FunctionEnd
-
-!macro NSIS_HOOK_PREINSTALL
-  ; Silent installs skip all pages. Also handle a fresh/custom install path.
-  Call EasyTierPrepareInstall
-!macroend
-
-!macro NSIS_HOOK_POSTINSTALL
-  !insertmacro EasyTierFinish
-!macroend
-
-!macro NSIS_HOOK_PREUNINSTALL
-  Call un.EasyTierPrepareUninstall
-!macroend
-
-!macro NSIS_HOOK_POSTUNINSTALL
-  !insertmacro EasyTierFinish
-!macroend

easytier-gui/src-tauri/windows/installer-processes.ps1

@@ -1,264 +0,0 @@
-param(
-    [ValidateSet('Prepare', 'Run')]
-    [string]$Action = 'Prepare',
-    [Parameter(Mandatory = $true)]
-    [string]$InstallDir,
-    [Parameter(Mandatory = $true)]
-    [int]$InstallerPid,
-    [int]$RestoreService = 1
-)
-
-$ErrorActionPreference = 'Stop'
-$eventName = "Local\EasyTierInstaller-$InstallerPid"
-$exePath = [IO.Path]::GetFullPath((Join-Path $InstallDir 'easytier-gui.exe')).Replace('\\?\', '')
-Add-Type -AssemblyName System.ServiceProcess
-
-# Process.MainModule and WMI cannot reliably read an elevated GUI's path from
-# the unelevated installer. Limited query access works across that boundary.
-Add-Type @'
-using System;
-using System.ComponentModel;
-using System.Runtime.InteropServices;
-using System.Text;
-
-public static class EasyTierInstallerProcess
-{
-    [DllImport("kernel32.dll", SetLastError = true)]
-    public static extern IntPtr OpenProcess(uint access, bool inherit, int pid);
-    [DllImport("kernel32.dll", CharSet = CharSet.Unicode, SetLastError = true)]
-    static extern bool QueryFullProcessImageName(IntPtr process, uint flags,
-        StringBuilder path, ref int size);
-    [DllImport("kernel32.dll", SetLastError = true)]
-    static extern bool TerminateProcess(IntPtr process, uint exitCode);
-    [DllImport("kernel32.dll")]
-    static extern uint WaitForSingleObject(IntPtr handle, uint milliseconds);
-    [DllImport("kernel32.dll")]
-    public static extern bool CloseHandle(IntPtr handle);
-
-    public static string Path(IntPtr process)
-    {
-        var path = new StringBuilder(32768);
-        int size = path.Capacity;
-        if (!QueryFullProcessImageName(process, 0, path, ref size))
-        {
-            int error = Marshal.GetLastWin32Error();
-            if (WaitForSingleObject(process, 0) == 0) return null;
-            throw new Win32Exception(error);
-        }
-        return path.ToString().Replace(@"\\?\", "");
-    }
-
-    public static bool Matches(int pid, string path, bool terminate)
-    {
-        // Keep the same handle for path verification, termination and waiting,
-        // so a recycled PID cannot cause us to terminate a different process.
-        IntPtr process = OpenProcess(terminate ? 0x101001u : 0x101000u, false, pid);
-        if (process == IntPtr.Zero)
-        {
-            int error = Marshal.GetLastWin32Error();
-            if (error == 87) return false; // The process has already exited.
-            throw new Win32Exception(error);
-        }
-        try
-        {
-            if (!String.Equals(Path(process), path, StringComparison.OrdinalIgnoreCase))
-                return false;
-            if (terminate)
-            {
-                if (!TerminateProcess(process, 0) && WaitForSingleObject(process, 0) != 0)
-                    throw new Win32Exception();
-                if (WaitForSingleObject(process, 30000) != 0)
-                    throw new TimeoutException("EasyTier did not exit within 30 seconds.");
-            }
-            return true;
-        }
-        finally { CloseHandle(process); }
-    }
-}
-'@
-
-function Get-AppProcesses {
-    foreach ($process in @(Get-Process -Name easytier-gui -ErrorAction SilentlyContinue)) {
-        if ([EasyTierInstallerProcess]::Matches($process.Id, $exePath, $false)) {
-            $process
-        }
-    }
-}
-
-function Get-AppService {
-    $service = Get-WmiObject -Class Win32_Service -Filter "Name='easytier-gui'"
-    if ($service) {
-        $path = [regex]::Match($service.PathName, '^(?:"([^"]+)"|(\S+))')
-        $binary = ($path.Groups[1].Value + $path.Groups[2].Value).Replace('\\?\', '')
-        if ($binary -eq $exePath) { $service }
-    }
-}
-
-function New-InstallerEvent([string]$name) {
-    $security = New-Object Security.AccessControl.EventWaitHandleSecurity
-    # RunAs may use another administrator account. The installer keeps its
-    # original identity; only these completion signals cross the boundary.
-    foreach ($sid in @(
-        [Security.Principal.WindowsIdentity]::GetCurrent().User,
-        (New-Object Security.Principal.SecurityIdentifier 'S-1-5-32-544')
-    )) {
-        $rule = New-Object Security.AccessControl.EventWaitHandleAccessRule(
-            $sid, 'FullControl', 'Allow')
-        $security.AddAccessRule($rule)
-    }
-    $created = $false
-    New-Object Threading.EventWaitHandle(
-        $false, [Threading.EventResetMode]::ManualReset, $name, [ref]$created, $security)
-}
-
-try {
-    if ($Action -eq 'Prepare') {
-        $service = Get-AppService
-        $runningService = $service -and $service.State -ne 'Stopped'
-        $needsElevation = $runningService
-        $processes = @()
-        try {
-            $processes = @(Get-AppProcesses)
-            foreach ($process in $processes) {
-                $handle = [EasyTierInstallerProcess]::OpenProcess(0x101001, $false, $process.Id)
-                if ($handle -eq [IntPtr]::Zero) { $needsElevation = $true }
-                else { [void][EasyTierInstallerProcess]::CloseHandle($handle) }
-            }
-        }
-        catch {
-            if ($_.Exception.GetBaseException().NativeErrorCode -ne 5) { throw }
-            $needsElevation = $true
-        }
-        if (!$runningService -and !$processes.Count -and !$needsElevation) {
-            Write-Output 0
-            exit 0
-        }
-
-        $ready = New-InstallerEvent "$eventName-ready"
-        $finish = New-InstallerEvent "$eventName-finish"
-        $done = New-InstallerEvent "$eventName-done"
-        $failed = New-InstallerEvent "$eventName-failed"
-        try {
-            $arguments = '-NoLogo -NoProfile -NonInteractive -ExecutionPolicy Bypass ' +
-                "-File `"$PSCommandPath`" -Action Run -InstallDir `"$InstallDir`" " +
-                "-InstallerPid $InstallerPid -RestoreService $RestoreService"
-            $start = @{
-                FilePath = "$PSHOME\powershell.exe"
-                ArgumentList = $arguments
-                WindowStyle = 'Hidden'
-                PassThru = $true
-            }
-            $identity = [Security.Principal.WindowsIdentity]::GetCurrent()
-            $principal = New-Object Security.Principal.WindowsPrincipal($identity)
-            if ($needsElevation -and !$principal.IsInRole([Security.Principal.WindowsBuiltInRole]::Administrator)) {
-                $start.Verb = 'RunAs'
-            }
-            $helper = Start-Process @start
-            while (!$ready.WaitOne(100)) {
-                if ($helper.HasExited) {
-                    throw 'Could not stop EasyTier.'
-                }
-            }
-            if ($failed.WaitOne(0)) { throw 'Could not stop EasyTier.' }
-            Write-Output $helper.Id
-        }
-        finally {
-            $ready.Dispose()
-            $finish.Dispose()
-            $done.Dispose()
-            $failed.Dispose()
-        }
-    }
-    else {
-        $ready = [Threading.EventWaitHandle]::OpenExisting("$eventName-ready")
-        $finish = [Threading.EventWaitHandle]::OpenExisting("$eventName-finish")
-        $done = [Threading.EventWaitHandle]::OpenExisting("$eventName-done")
-        $failed = [Threading.EventWaitHandle]::OpenExisting("$eventName-failed")
-        $installer = [Diagnostics.Process]::GetProcessById($InstallerPid)
-        $installerExit = New-Object Threading.ManualResetEvent($false)
-        $installerExit.SafeWaitHandle = New-Object Microsoft.Win32.SafeHandles.SafeWaitHandle(
-            $installer.Handle, $false)
-        $restart = $false
-        $controller = $null
-        $serviceProcess = $null
-        $startMode = $null
-        try {
-            $service = Get-AppService
-            if ($service -and $service.State -ne 'Stopped') {
-                $controller = New-Object ServiceProcess.ServiceController('easytier-gui')
-                $restart = $RestoreService -ne 0 -and $service.State -ne 'Stop Pending' -and
-                    $service.StartMode -ne 'Disabled'
-                if ($service.StartMode -ne 'Disabled') {
-                    # Released versions can exit without reporting Stopped,
-                    # which queues SCM recovery even after an explicit stop.
-                    # Keep recovery from starting the old binary during setup.
-                    $originalStartMode = $service.StartMode
-                    $result = $service.ChangeStartMode('Disabled').ReturnValue
-                    if ($result -ne 0) { throw "Could not disable service startup: $result" }
-                    $startMode = if ($originalStartMode -eq 'Auto') { 'Automatic' } else { 'Manual' }
-                }
-                if ($service.ProcessId) {
-                    $serviceProcess = [Diagnostics.Process]::GetProcessById($service.ProcessId)
-                    # Acquire the handle before stopping, while the PID still exists.
-                    $null = $serviceProcess.Handle
-                }
-                if ($controller.Status -eq 'StartPending') {
-                    $controller.WaitForStatus('Running', [TimeSpan]::FromSeconds(30))
-                }
-                if ($controller.Status -ne 'StopPending') { $controller.Stop() }
-                $controller.WaitForStatus('Stopped', [TimeSpan]::FromSeconds(30))
-                if ($serviceProcess -and !$serviceProcess.WaitForExit(30000)) {
-                    throw 'The EasyTier service process did not exit within 30 seconds.'
-                }
-            }
-            foreach ($process in @(Get-AppProcesses)) {
-                [void][EasyTierInstallerProcess]::Matches($process.Id, $exePath, $true)
-            }
-            [void]$ready.Set()
-            [void][Threading.WaitHandle]::WaitAny(@($finish, $installerExit))
-        }
-        finally {
-            # Also restore after cancellation/failure, provided the executable
-            # still exists. A completed uninstall must not restart the service.
-            try {
-                if ($startMode) {
-                    $result = $service.ChangeStartMode($startMode).ReturnValue
-                    if ($result -ne 0) { throw "Could not restore service startup: $result" }
-                }
-                if ($restart -and (Test-Path -LiteralPath $exePath)) {
-                    $controller.Refresh()
-                    if ($controller.Status -eq 'StopPending') {
-                        $controller.WaitForStatus('Stopped', [TimeSpan]::FromSeconds(30))
-                    }
-                    if ($controller.Status -eq 'Stopped') {
-                        $controller.Start()
-                        $controller.WaitForStatus('Running', [TimeSpan]::FromSeconds(30))
-                    }
-                }
-            }
-            finally {
-                if ($controller) { $controller.Dispose() }
-                if ($serviceProcess) { $serviceProcess.Dispose() }
-                $installerExit.Dispose()
-                $installer.Dispose()
-            }
-        }
-    }
-}
-catch {
-    if ($Action -eq 'Run' -and $failed) { [void]$failed.Set() }
-    Write-Error $_ -ErrorAction Continue
-    exit 1
-}
-finally {
-    if ($Action -eq 'Run' -and $done) {
-        # Completion events remain accessible to the original user even when
-        # UAC ran the helper under a different administrator account.
-        [void]$done.Set()
-        [void]$ready.Set()
-        $ready.Dispose()
-        $finish.Dispose()
-        $done.Dispose()
-        $failed.Dispose()
-    }
-}

easytier-gui/src-tauri/windows/test-installer-processes.ps1

@@ -1,169 +0,0 @@
-# Run with Windows PowerShell as administrator on a machine without an
-# installed easytier-gui service. All processes and files are test fixtures.
-# Use -GuiOnly from an unelevated shell to test the path that needs no UAC.
-param([switch]$GuiOnly)
-
-$ErrorActionPreference = 'Stop'
-$helperScript = Join-Path $PSScriptRoot 'installer-processes.ps1'
-$testDir = Join-Path $env:TEMP ("EasyTier installer test " + [guid]::NewGuid())
-$appDir = Join-Path $testDir 'app'
-$otherDir = Join-Path $testDir 'other'
-$exe = Join-Path $appDir 'easytier-gui.exe'
-$helper = $null
-$gui = $null
-$other = $null
-$parent = $null
-$createdService = $false
-
-function Assert($condition, [string]$message) {
-    if (!$condition) { throw $message }
-}
-
-function Prepare([int]$parentPid = $PID, [int]$restore = 1) {
-    $output = & "$PSHOME\powershell.exe" -NoProfile -ExecutionPolicy Bypass `
-        -File $helperScript -InstallDir $appDir -InstallerPid $parentPid -RestoreService $restore
-    Assert ($LASTEXITCODE -eq 0) "Prepare failed: $output"
-    $helperPid = [int]($output | Select-Object -Last 1)
-    if ($helperPid) {
-        $process = [Diagnostics.Process]::GetProcessById($helperPid)
-        $null = $process.Handle
-        return $process
-    }
-}
-
-function Finish($process) {
-    if (!$process) { return }
-    $finish = [Threading.EventWaitHandle]::OpenExisting("Local\EasyTierInstaller-$PID-finish")
-    $done = [Threading.EventWaitHandle]::OpenExisting("Local\EasyTierInstaller-$PID-done")
-    $failed = [Threading.EventWaitHandle]::OpenExisting("Local\EasyTierInstaller-$PID-failed")
-    try {
-        [void]$finish.Set()
-        Assert ($done.WaitOne(65000)) 'The helper did not signal completion.'
-        Assert (!$failed.WaitOne(0)) 'The helper reported a failure.'
-    }
-    finally {
-        $finish.Dispose()
-        $done.Dispose()
-        $failed.Dispose()
-    }
-    Assert ($process.WaitForExit(5000)) 'The helper did not finish.'
-    Assert ($process.ExitCode -eq 0) "The helper failed: $($process.ExitCode)"
-    $process.Dispose()
-}
-
-function Start-TestService {
-    Start-Service easytier-gui
-    (Get-Service easytier-gui).WaitForStatus('Running', [TimeSpan]::FromSeconds(10))
-}
-
-if (Get-Service easytier-gui -ErrorAction SilentlyContinue) {
-    throw 'An easytier-gui service already exists; refusing to touch it.'
-}
-try {
-    [void](New-Item -ItemType Directory -Path $appDir, $otherDir)
-    Add-Type -ReferencedAssemblies System.ServiceProcess -OutputAssembly $exe -OutputType ConsoleApplication @'
-using System;
-using System.ServiceProcess;
-using System.Threading;
-
-class FixtureService : ServiceBase
-{
-    static bool legacy;
-    public FixtureService() { ServiceName = "easytier-gui"; }
-    protected override void OnStop()
-    {
-        // Model released EasyTier versions that exit without reporting Stopped.
-        if (legacy) Environment.Exit(0);
-    }
-    static void Main(string[] args)
-    {
-        if (args.Length > 0)
-        {
-            legacy = args[0] == "--legacy";
-            ServiceBase.Run(new FixtureService());
-        }
-        else Thread.Sleep(Timeout.Infinite);
-    }
-}
-'@
-    Copy-Item $exe (Join-Path $otherDir 'easytier-gui.exe')
-
-    $helper = Prepare
-    Assert (!$helper) 'An idle installation must not start a helper.'
-    Write-Host 'PASS: no running process or service'
-
-    $gui = Start-Process $exe -PassThru
-    $other = Start-Process (Join-Path $otherDir 'easytier-gui.exe') -PassThru
-    $helper = Prepare
-    Assert $gui.HasExited 'The GUI was not stopped.'
-    Assert (!$other.HasExited) 'A GUI from a different installation was stopped.'
-    Finish $helper
-    $helper = $null
-    Write-Host 'PASS: close GUI, preserve a different installation'
-    if ($GuiOnly) { return }
-
-    New-Service -Name easytier-gui -BinaryPathName "`"$exe`" --legacy" -StartupType Automatic | Out-Null
-    $createdService = $true
-    & "$env:SystemRoot\System32\sc.exe" failure easytier-gui reset= 0 actions= restart/1000 | Out-Null
-    Assert ($LASTEXITCODE -eq 0) 'Could not configure service recovery.'
-    Start-TestService
-    $gui = Start-Process $exe -PassThru
-    $helper = Prepare
-    Start-Sleep -Milliseconds 1500
-    Assert ((Get-Service easytier-gui).Status -eq 'Stopped') 'The service was not stopped.'
-    Assert ((Get-WmiObject Win32_Service -Filter "Name='easytier-gui'").StartMode -eq 'Disabled') 'Service recovery was not inhibited.'
-    Assert $gui.HasExited 'The GUI alongside the service was not stopped.'
-    Move-Item $exe "$exe.old"
-    Copy-Item "$exe.old" $exe
-    Finish $helper
-    $helper = $null
-    Assert ((Get-Service easytier-gui).Status -eq 'Running') 'The running service was not restored.'
-    Assert ((Get-WmiObject Win32_Service -Filter "Name='easytier-gui'").StartMode -eq 'Auto') 'Automatic startup was not restored.'
-    Write-Host 'PASS: stop legacy service despite failure recovery, replace executable, restore service'
-
-    Set-Service easytier-gui -StartupType Disabled
-    Stop-Service easytier-gui
-    (Get-Service easytier-gui).WaitForStatus('Stopped', [TimeSpan]::FromSeconds(10))
-    Start-Sleep -Milliseconds 1500
-    Set-Service easytier-gui -StartupType Manual
-    $helper = Prepare
-    Assert (!$helper) 'A stopped service must not start a helper.'
-    $gui = Start-Process $exe -PassThru
-    $helper = Prepare
-    Finish $helper
-    $helper = $null
-    Assert ((Get-Service easytier-gui).Status -eq 'Stopped') 'A previously stopped service was started.'
-    Write-Host 'PASS: preserve stopped service state'
-
-    Start-TestService
-    $parent = Start-Process "$PSHOME\powershell.exe" -ArgumentList '-NoProfile -Command Start-Sleep 300' -WindowStyle Hidden -PassThru
-    $helper = Prepare $parent.Id
-    Assert ((Get-Service easytier-gui).Status -eq 'Stopped') 'The service was not stopped before cancellation.'
-    $parent.Kill()
-    Assert ($helper.WaitForExit(35000)) 'The helper did not handle installer exit.'
-    Assert ($helper.ExitCode -eq 0) 'The helper failed while handling installer exit.'
-    $helper.Dispose()
-    $helper = $null
-    Assert ((Get-Service easytier-gui).Status -eq 'Running') 'Cancellation did not restore the service.'
-    Assert ((Get-WmiObject Win32_Service -Filter "Name='easytier-gui'").StartMode -eq 'Manual') 'Manual startup was not restored.'
-    Write-Host 'PASS: restore service when the installer exits'
-
-    $helper = Prepare $PID 0
-    Finish $helper
-    $helper = $null
-    Assert ((Get-Service easytier-gui).Status -eq 'Stopped') 'Uninstallation restarted the service.'
-    Write-Host 'PASS: do not restart the service after uninstall'
-}
-finally {
-    if ($helper -and !$helper.HasExited) { Finish $helper }
-    if ($createdService) {
-        Set-Service easytier-gui -StartupType Disabled
-        Stop-Service easytier-gui -ErrorAction SilentlyContinue
-        & "$env:SystemRoot\System32\sc.exe" delete easytier-gui | Out-Null
-    }
-    foreach ($process in @($gui, $other, $parent)) {
-        if ($process -and !$process.HasExited) { $process.Kill(); $process.WaitForExit() }
-        if ($process) { $process.Dispose() }
-    }
-    Remove-Item -LiteralPath $testDir -Recurse -Force
-}

easytier-web/src/client_manager/mod.rs

@@ -347,11 +347,7 @@ impl ClientManager {
         ) && let Some(session) = self.get_session_by_machine_id(user_id, &machine_id)
         {
             session
-                .notify_managed_runtime_state_changed(
-                    user_id,
-                    machine_id,
-                    config_revision.as_deref(),
-                )
+                .notify_managed_runtime_state_changed(user_id, machine_id)
                 .await;
         }
         Ok(())
@@ -396,17 +392,13 @@ impl ClientManager {
                 machine_id,
                 ManagedConfigPersistedChange {
                     expected_revision: expected_config_revision,
-                    target_revision: config_revision.clone(),
+                    target_revision: config_revision,
                     dirty_instance_ids,
                 },
             );
             if changed && let Some(session) = self.get_session_by_machine_id(user_id, &machine_id) {
                 session
-                    .notify_managed_runtime_state_changed(
-                        user_id,
-                        machine_id,
-                        Some(&config_revision),
-                    )
+                    .notify_managed_runtime_state_changed(user_id, machine_id)
                     .await;
             }
         }
@@ -424,7 +416,7 @@ impl ClientManager {
             && let Some(session) = self.get_session_by_machine_id(user_id, &machine_id)
         {
             session
-                .notify_managed_runtime_state_changed(user_id, machine_id, None)
+                .notify_managed_runtime_state_changed(user_id, machine_id)
                 .await;
         }
     }
@@ -659,7 +651,6 @@ mod tests {
     #[derive(Debug, Clone)]
     struct TestWebhookState {
         validate_responses: Arc<tokio::sync::Mutex<VecDeque<bool>>>,
-        config_revision: Arc<tokio::sync::RwLock<String>>,
         validate_count: Arc<AtomicUsize>,
         block_second_validate: Arc<AtomicBool>,
         allow_second_validate: Arc<AtomicBool>,
@@ -674,7 +665,6 @@ mod tests {
                 validate_responses: Arc::new(tokio::sync::Mutex::new(
                     validate_responses.into_iter().collect(),
                 )),
-                config_revision: Arc::new(tokio::sync::RwLock::new("rev-initial".to_owned())),
                 validate_count: Arc::new(AtomicUsize::new(0)),
                 block_second_validate: Arc::new(AtomicBool::new(false)),
                 allow_second_validate: Arc::new(AtomicBool::new(true)),
@@ -738,8 +728,8 @@ mod tests {
 
         Json(json!({
             "valid": true,
-            "binding_version": 1,
-            "config_revision": state.config_revision.read().await.clone()
+            "binding_version": count,
+            "config_revision": format!("validated-rev-{count}")
         }))
     }
 
@@ -1280,8 +1270,7 @@ mod tests {
 
     #[tokio::test]
     async fn compact_runtime_preserves_unsupported_web_config_during_hot_patch() {
-        let (webhook_config, webhook_server, webhook_state) = test_webhook_config().await;
-        *webhook_state.config_revision.write().await = "compact-initial".to_owned();
+        let (webhook_config, webhook_server, _) = test_webhook_config().await;
         let mut mgr = ClientManager::new(
             Db::memory_db().await,
             None,
@@ -1318,7 +1307,6 @@ mod tests {
         })
         .await;
 
-        *webhook_state.config_revision.write().await = "compact-patched".to_owned();
         mgr.reconcile_managed_network_configs(
             user_id,
             machine_id,
@@ -1346,284 +1334,9 @@ mod tests {
         webhook_server.abort();
     }
 
-    #[tokio::test]
-    async fn receiver_takeover_preserves_running_core_until_config_arrives() {
-        for stale_config in [false, true] {
-            let (old_webhook, old_server, _) = test_webhook_config().await;
-            let mut old = ClientManager::new(
-                Db::memory_db().await,
-                None,
-                HeartbeatPolicy::from_millis(1_000, 15_000).unwrap(),
-                Arc::new(FeatureFlags::default()),
-                old_webhook,
-            );
-            let old_addr = add_random_udp_listener(&mut old).await;
-            let machine_id = uuid::Uuid::new_v4();
-            let instance_id = uuid::Uuid::new_v4();
-            let old_user = old
-                .db()
-                .auto_create_user(MANAGED_CONFIG_TOKEN)
-                .await
-                .unwrap()
-                .id;
-            let desired = initial_managed_network_config(instance_id);
-            old.reconcile_managed_network_configs(
-                old_user,
-                machine_id,
-                vec![managed_config(instance_id, desired.clone())],
-                Some("rev-initial".to_owned()),
-                None,
-            )
-            .await
-            .unwrap();
-            let core = Arc::new(native_instance_manager());
-            let old_client = start_web_client_for_test(old_addr, machine_id, core.clone()).await;
-            wait_for_applied_revision(&old, old_user, machine_id, "rev-initial").await;
-            let original = core.instance(instance_id).unwrap();
-            drop(old_client);
-
-            let state = TestWebhookState::with_blocked_connected([true]);
-            let (webhook, server, state) = test_webhook_config_with_state(state).await;
-            let mut replacement = ClientManager::new(
-                Db::memory_db().await,
-                None,
-                HeartbeatPolicy::from_millis(1_000, 15_000).unwrap(),
-                Arc::new(FeatureFlags::default()),
-                webhook,
-            );
-            let addr = add_random_udp_listener(&mut replacement).await;
-            let user_id = replacement
-                .db()
-                .auto_create_user(MANAGED_CONFIG_TOKEN)
-                .await
-                .unwrap()
-                .id;
-            if stale_config {
-                let obsolete_id = uuid::Uuid::new_v4();
-                replacement
-                    .reconcile_managed_network_configs(
-                        user_id,
-                        machine_id,
-                        vec![managed_config(
-                            obsolete_id,
-                            initial_managed_network_config(obsolete_id),
-                        )],
-                        Some("outdated".to_owned()),
-                        None,
-                    )
-                    .await
-                    .unwrap();
-            }
-            let _client = start_web_client_for_test(addr, machine_id, core.clone()).await;
-            wait_for_condition(
-                || async { state.connected_count() == 1 },
-                Duration::from_secs(5),
-            )
-            .await;
-            let session = replacement
-                .get_session_by_machine_id(user_id, &machine_id)
-                .unwrap();
-
-            // Authentication and real heartbeats continue while publication is
-            // withheld. Neither an empty DB nor an obsolete Exact Set owns the
-            // already running Core instance.
-            for _ in 0..3 {
-                let before = session.get_heartbeat_req().await.unwrap().report_time;
-                wait_for_condition(
-                    || async { session.get_heartbeat_req().await.unwrap().report_time != before },
-                    Duration::from_secs(3),
-                )
-                .await;
-                assert_eq!(core.instance_ids(), vec![instance_id]);
-                assert!(Arc::ptr_eq(&original, &core.instance(instance_id).unwrap()));
-                assert!(session.applied_config_revision().await.is_none());
-            }
-
-            state.allow_connected();
-            // Registration must wake validation before its ten-heartbeat
-            // periodic cadence, without repeatedly registering the same binding.
-            wait_for_condition(
-                || async { state.validate_count() == 2 },
-                Duration::from_secs(2),
-            )
-            .await;
-            assert_eq!(state.connected_count(), 1);
-            assert!(Arc::ptr_eq(&original, &core.instance(instance_id).unwrap()));
-
-            replacement
-                .reconcile_managed_network_configs(
-                    user_id,
-                    machine_id,
-                    vec![managed_config(instance_id, desired)],
-                    Some("rev-initial".to_owned()),
-                    None,
-                )
-                .await
-                .unwrap();
-            wait_for_applied_revision(&replacement, user_id, machine_id, "rev-initial").await;
-            assert_eq!(core.instance_ids(), vec![instance_id]);
-            assert!(Arc::ptr_eq(&original, &core.instance(instance_id).unwrap()));
-
-            // A versioned authoritative empty set still removes web-owned
-            // instances and keeps the authenticated management session alive.
-            *state.config_revision.write().await = "empty".to_owned();
-            replacement
-                .reconcile_managed_network_configs(
-                    user_id,
-                    machine_id,
-                    Vec::new(),
-                    Some("empty".to_owned()),
-                    Some("rev-initial".to_owned()),
-                )
-                .await
-                .unwrap();
-            wait_for_applied_revision(&replacement, user_id, machine_id, "empty").await;
-            assert!(core.instance_ids().is_empty());
-            assert!(
-                replacement
-                    .get_session_by_machine_id(user_id, &machine_id)
-                    .is_some()
-            );
-            server.abort();
-            old_server.abort();
-        }
-    }
-
-    #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
-    #[ignore = "1000 real management connections; run explicitly with a raised file limit"]
-    async fn receiver_takeover_wave_allows_an_independent_mount() {
-        use futures::StreamExt;
-
-        let (old_webhook, old_server, _) = test_webhook_config().await;
-        let mut old = ClientManager::new(
-            Db::memory_db().await,
-            None,
-            HeartbeatPolicy::default(),
-            Arc::new(FeatureFlags::default()),
-            old_webhook,
-        );
-        let old_addr = add_random_udp_listener(&mut old).await;
-        let (webhook, server, state) = test_webhook_config().await;
-        let mut replacement = ClientManager::new(
-            Db::memory_db().await,
-            None,
-            HeartbeatPolicy::default(),
-            Arc::new(FeatureFlags::default()),
-            webhook,
-        );
-        let addr = add_random_udp_listener(&mut replacement).await;
-        // Provision the account once: password hashing is unrelated to a wave
-        // of existing devices reconnecting to another receiver.
-        let old_user = old
-            .db()
-            .auto_create_user(MANAGED_CONFIG_TOKEN)
-            .await
-            .unwrap()
-            .id;
-        let user_id = replacement
-            .db()
-            .auto_create_user(MANAGED_CONFIG_TOKEN)
-            .await
-            .unwrap()
-            .id;
-        let devices: Vec<_> = (0..1_000)
-            .map(|_| {
-                (
-                    uuid::Uuid::new_v4(),
-                    Arc::new(native_compact_instance_manager_with_runtime(
-                        tokio::runtime::Handle::current(),
-                    )),
-                )
-            })
-            .collect();
-        let old_clients: Vec<_> = futures::stream::iter(&devices)
-            .map(|(machine_id, core)| async {
-                old.reconcile_managed_network_configs(
-                    old_user,
-                    *machine_id,
-                    Vec::new(),
-                    Some("rev-initial".to_owned()),
-                    None,
-                )
-                .await
-                .unwrap();
-                let client = start_web_client_for_test(old_addr, *machine_id, core.clone()).await;
-                wait_for_applied_revision(&old, old_user, *machine_id, "rev-initial").await;
-                client
-            })
-            .buffer_unordered(64)
-            .collect()
-            .await;
-
-        let mount_machine = uuid::Uuid::new_v4();
-        let mount_core = Arc::new(native_compact_instance_manager_with_runtime(
-            tokio::runtime::Handle::current(),
-        ));
-        let _mount_client =
-            start_web_client_for_test(addr, mount_machine, mount_core.clone()).await;
-        wait_for_validated_user(&replacement, mount_machine).await;
-        drop(old_clients);
-        let started = tokio::time::Instant::now();
-        let takeover = futures::stream::iter(&devices)
-            .map(|(machine_id, core)| async {
-                let client = start_web_client_for_test(addr, *machine_id, core.clone()).await;
-                wait_for_validated_user(&replacement, *machine_id).await;
-                replacement
-                    .reconcile_managed_network_configs(
-                        user_id,
-                        *machine_id,
-                        Vec::new(),
-                        Some("rev-initial".to_owned()),
-                        None,
-                    )
-                    .await
-                    .unwrap();
-                wait_for_applied_revision(&replacement, user_id, *machine_id, "rev-initial").await;
-                client
-            })
-            .buffer_unordered(64)
-            .collect::<Vec<_>>();
-        let mount = async {
-            wait_for_condition(
-                || async { state.connected_count() >= 11 },
-                Duration::from_secs(12),
-            )
-            .await;
-            let started = tokio::time::Instant::now();
-            let instance_id = uuid::Uuid::new_v4();
-            replacement
-                .reconcile_managed_network_configs(
-                    user_id,
-                    mount_machine,
-                    vec![managed_config(
-                        instance_id,
-                        initial_managed_network_config(instance_id),
-                    )],
-                    Some("rev-initial".to_owned()),
-                    None,
-                )
-                .await
-                .unwrap();
-            wait_for_applied_revision(&replacement, user_id, mount_machine, "rev-initial").await;
-            assert_eq!(mount_core.instance_ids(), vec![instance_id]);
-            started.elapsed()
-        };
-        let (clients, mount_elapsed) = tokio::join!(takeover, mount);
-        assert_eq!(clients.len(), 1_000);
-        assert_eq!(replacement.list_sessions().await.len(), 1_001);
-        println!(
-            "1000 receiver takeovers: {:?}; independent mount: {:?}; validations: {}",
-            started.elapsed(),
-            mount_elapsed,
-            state.validate_count()
-        );
-        server.abort();
-        old_server.abort();
-    }
-
     #[tokio::test]
     async fn managed_web_config_revision_updates_running_core_config() {
-        let (webhook_config, webhook_server, webhook_state) = test_webhook_config().await;
+        let (webhook_config, webhook_server, _) = test_webhook_config().await;
         let mut mgr = ClientManager::new(
             Db::memory_db().await,
             None,
@@ -1687,7 +1400,6 @@ mod tests {
 
         // Online revision update: web-owned running config is fully overwritten
         // when non-hot-patch flags such as enable_kcp_proxy change.
-        *webhook_state.config_revision.write().await = "rev-updated".to_owned();
         mgr.reconcile_managed_network_configs(
             user_id,
             machine_id,
@@ -1742,7 +1454,6 @@ mod tests {
                 .is_none()
         );
 
-        *webhook_state.config_revision.write().await = "rev-webhook-redelivery".to_owned();
         mgr.reconcile_managed_network_configs(
             user_id,
             machine_id,

easytier-web/src/client_manager/session.rs

@@ -133,13 +133,9 @@ pub struct SessionData {
 
     storage_token: Option<StorageToken>,
     binding_version: Option<u64>,
-    // Authentication does not prove that this receiver has the desired Exact Set.
-    // Keep that authority on the Session, not in runtime continuity shared with
-    // an earlier connection.
-    webhook_config_revision: Option<String>,
     managed_runtime: SharedManagedRuntimeState,
     direct_run_failed_instance_ids: HashSet<String>,
-    notifier: broadcast::Sender<()>,
+    notifier: broadcast::Sender<HeartbeatRequest>,
     req: Option<HeartbeatRequest>,
     location: Option<Location>,
     heartbeat_count: std::sync::atomic::AtomicU32,
@@ -169,7 +165,6 @@ impl SessionData {
             client_url,
             storage_token: None,
             binding_version: None,
-            webhook_config_revision: None,
             managed_runtime: Arc::new(Mutex::new(ManagedRuntimeState::default())),
             direct_run_failed_instance_ids: HashSet::new(),
             notifier: tx,
@@ -190,22 +185,10 @@ impl SessionData {
         self.req.clone()
     }
 
-    pub fn heartbeat_waiter(&self) -> broadcast::Receiver<()> {
+    pub fn heartbeat_waiter(&self) -> broadcast::Receiver<HeartbeatRequest> {
         self.notifier.subscribe()
     }
 
-    fn runtime_config_is_ready(&self, persisted_revision: Option<&str>) -> bool {
-        if !self.webhook_config.is_enabled() {
-            return true;
-        }
-        match self.webhook_config_revision.as_deref() {
-            // Standalone webhook integrations can omit both managed fields.
-            Some("") if self.binding_version == Some(0) => true,
-            Some(revision) if !revision.is_empty() => persisted_revision == Some(revision),
-            _ => false,
-        }
-    }
-
     pub fn location(&self) -> Option<&Location> {
         self.location.as_ref()
     }
@@ -319,11 +302,6 @@ async fn record_webhook_connected_binding_if_current(
     binding_version: u64,
 ) -> Option<ConnectedBindingRecord> {
     let session_data = session_data.upgrade()?;
-    let storage = Storage::try_from(session_data.read().await.storage.clone()).ok()?;
-    let persisted_revision = storage
-        .db()
-        .get_managed_config_revision((storage_token.user_id, storage_token.machine_id))
-        .await;
     let mut data = session_data.write().await;
     if !connection_state_matches(&data, storage_token, binding_version) {
         return Some(ConnectedBindingRecord::IdentityStale);
@@ -335,18 +313,7 @@ async fn record_webhook_connected_binding_if_current(
     {
         return Some(ConnectedBindingRecord::OwnershipLost);
     }
-    let newly_connected = data.webhook_connected_binding_version != Some(binding_version);
     data.webhook_connected_binding_version = Some(binding_version);
-    if newly_connected
-        && data.webhook_config.is_enabled()
-        && !persisted_revision
-            .as_ref()
-            .is_ok_and(|revision| data.runtime_config_is_ready(revision.as_deref()))
-    {
-        // The connected ACK makes the receiver route publishable. Retry the
-        // existing validation path now, rather than waiting ten heartbeats.
-        SessionRpcService::mark_webhook_validation_state_changed_locked(&mut data).notify_one();
-    }
     Some(ConnectedBindingRecord::Recorded)
 }
 
@@ -781,13 +748,13 @@ impl SessionRpcService {
                     data.session_epoch,
                 );
             }
-            let runtime_notify =
-                (authorized && data.storage_token.is_some()).then(|| data.notifier.clone());
+            let runtime_notify = (authorized && data.storage_token.is_some())
+                .then(|| (data.notifier.clone(), runtime_req));
             (notify, runtime_notify)
         };
 
-        if let Some(notifier) = runtime_notify {
-            let _ = notifier.send(());
+        if let Some((notifier, runtime_req)) = runtime_notify {
+            let _ = notifier.send(runtime_req);
         }
         if let Some(notify) = notify {
             notify.notify_one();
@@ -892,7 +859,7 @@ impl SessionRpcService {
 
         let report_time = Self::heartbeat_report_timestamp(&runtime_req);
         storage.update_session_client(storage_token, report_time, true, session_epoch);
-        let _ = notifier.send(());
+        let _ = notifier.send(runtime_req);
         if let Some(notify) = validation_notify {
             notify.notify_one();
         }
@@ -1081,10 +1048,9 @@ impl Session {
         &self,
         user_id: i32,
         machine_id: uuid::Uuid,
-        published_revision: Option<&str>,
     ) {
         let notify = {
-            let mut data = self.data.write().await;
+            let data = self.data.read().await;
             if !data.auth_state.is_authorized() {
                 return;
             }
@@ -1095,21 +1061,10 @@ impl Session {
             {
                 return;
             }
-            if let Some(revision) = published_revision
-                && data.webhook_config.is_enabled()
-                && data.webhook_config_revision.as_deref() != Some(revision)
-            {
-                // Only validation establishes the desired revision. A delayed
-                // publication callback must not replace newer authority, even
-                // when its revision is still the receiver's persisted value.
-                // Fence validation already in flight and request current authority.
-                SessionRpcService::mark_webhook_validation_state_changed_locked(&mut data)
-                    .notify_one();
-            }
-            data.req.as_ref().map(|_| data.notifier.clone())
+            data.req.clone().map(|req| (data.notifier.clone(), req))
         };
-        if let Some(notifier) = notify {
-            let _ = notifier.send(());
+        if let Some((notifier, req)) = notify {
+            let _ = notifier.send(req);
         }
     }
 
@@ -1127,10 +1082,10 @@ impl Session {
             runtime.runtime_config_epoch = runtime.runtime_config_epoch.wrapping_add(1);
             runtime.runtime_config_cache_epoch = runtime.runtime_config_cache_epoch.wrapping_add(1);
             drop(runtime);
-            data.req.as_ref().map(|_| data.notifier.clone())
+            data.req.clone().map(|req| (data.notifier.clone(), req))
         };
-        if let Some(notifier) = notify {
-            let _ = notifier.send(());
+        if let Some((notifier, req)) = notify {
+            let _ = notifier.send(req);
         }
     }
 
@@ -1850,7 +1805,7 @@ mod tests {
         assert_eq!(data.binding_version, Some(1));
         assert_eq!(data.webhook_connected_binding_version, Some(1));
         drop(data);
-        heartbeat_waiter.recv().await.unwrap();
+        assert_eq!(heartbeat_waiter.recv().await.unwrap().user_token, "token");
         assert!(
             storage
                 .db()

easytier-web/src/client_manager/session/runtime_revision.rs

@@ -24,12 +24,11 @@ use crate::client_manager::{
 };
 
 async fn recv_latest_heartbeat(
-    session_data: &std::sync::Weak<RwLock<SessionData>>,
-    heartbeat_waiter: &mut broadcast::Receiver<()>,
+    heartbeat_waiter: &mut broadcast::Receiver<HeartbeatRequest>,
 ) -> Option<HeartbeatRequest> {
-    loop {
+    let mut req = loop {
         match heartbeat_waiter.recv().await {
-            Ok(()) => break,
+            Ok(req) => break req,
             Err(broadcast::error::RecvError::Lagged(skipped)) => {
                 tracing::warn!(
                     skipped,
@@ -41,23 +40,21 @@ async fn recv_latest_heartbeat(
                 return None;
             }
         }
-    }
+    };
 
     // Drop any heartbeat backlog accumulated while the previous reconcile
     // round was doing DB/RPC IO. The newest heartbeat has the freshest
     // runtime instance list, which is all this task needs.
     loop {
         match heartbeat_waiter.try_recv() {
-            Ok(()) => {}
+            Ok(next_req) => req = next_req,
             Err(broadcast::error::TryRecvError::Empty) => break,
             Err(broadcast::error::TryRecvError::Lagged(_)) => continue,
             Err(broadcast::error::TryRecvError::Closed) => return None,
         }
     }
 
-    // Notifications can be delayed by validation or publication IO. Read the
-    // current heartbeat instead of replaying the snapshot taken before that IO.
-    session_data.upgrade()?.read().await.req()
+    Some(req)
 }
 
 struct SessionClients {
@@ -68,7 +65,7 @@ struct SessionClients {
 
 pub(super) async fn reconcile_network_configs_on_heartbeat(
     session_data: std::sync::Weak<RwLock<SessionData>>,
-    mut heartbeat_waiter: broadcast::Receiver<()>,
+    mut heartbeat_waiter: broadcast::Receiver<HeartbeatRequest>,
     storage: WeakRefStorage,
     rpc_client: SessionRpcClient,
     config_client: SessionConfigClient,
@@ -81,7 +78,7 @@ pub(super) async fn reconcile_network_configs_on_heartbeat(
     };
     let mut cache = ReconcileCache::default();
     loop {
-        let Some(req) = recv_latest_heartbeat(&session_data, &mut heartbeat_waiter).await else {
+        let Some(req) = recv_latest_heartbeat(&mut heartbeat_waiter).await else {
             return;
         };
         let Some(storage) = storage.upgrade() else {
@@ -407,15 +404,6 @@ async fn prepare_reconcile_round(
             RoundStatus::Skip => return RoundStatus::Skip,
             RoundStatus::Stop => return RoundStatus::Stop,
         };
-    {
-        let Some(data) = session_data.upgrade() else {
-            return RoundStatus::Stop;
-        };
-        let data = data.read().await;
-        if !data.runtime_config_is_ready(target_config_revision.as_deref()) {
-            return RoundStatus::Skip;
-        }
-    }
     let should_apply_runtime_revision =
         target_config_revision.is_some() && target_config_revision != applied_config_revision;
     let mut scope = if should_apply_runtime_revision {
@@ -898,9 +886,7 @@ async fn begin_managed_runtime_mutation(
         return false;
     };
     let data = data.write().await;
-    if !SessionRpcService::runtime_heartbeat_is_current_locked(&data, &round.req)
-        || !data.runtime_config_is_ready(round.target_config_revision.as_deref())
-    {
+    if !SessionRpcService::runtime_heartbeat_is_current_locked(&data, &round.req) {
         return false;
     }
     let managed_runtime = data.managed_runtime.clone();
@@ -1279,9 +1265,7 @@ async fn mark_config_revision_applied_if_current(
     };
     let notify = {
         let mut data = data.write().await;
-        if !SessionRpcService::runtime_heartbeat_is_current_locked(&data, &round.req)
-            || !data.runtime_config_is_ready(round.target_config_revision.as_deref())
-        {
+        if !SessionRpcService::runtime_heartbeat_is_current_locked(&data, &round.req) {
             return RoundStatus::Ready(());
         }
         record_applied_config_revision(
@@ -1335,286 +1319,6 @@ mod tests {
 
     use super::*;
 
-    async fn takeover_session(
-        persisted_revision: Option<&str>,
-    ) -> (
-        crate::client_manager::storage::Storage,
-        std::sync::Arc<RwLock<SessionData>>,
-        ReconcileRound,
-    ) {
-        use super::super::{SessionAuthState, webhook_validation};
-        use crate::client_manager::storage::{Storage, StorageToken};
-
-        let storage = Storage::new(crate::db::Db::memory_db().await);
-        let user_id = storage.db().auto_create_user("token").await.unwrap().id;
-        let machine_id = uuid::Uuid::new_v4();
-        if let Some(revision) = persisted_revision {
-            storage
-                .db()
-                .set_managed_config_revision((user_id, machine_id), revision)
-                .await
-                .unwrap();
-        }
-        let req = HeartbeatRequest {
-            user_token: "token".to_owned(),
-            machine_id: Some(machine_id.into()),
-            running_network_instances: vec![uuid::Uuid::new_v4().into()],
-            ..Default::default()
-        };
-        let client_url = "tcp://127.0.0.1:1001".parse().unwrap();
-        let webhook = std::sync::Arc::new(crate::webhook::WebhookConfig::new(
-            Some("http://127.0.0.1:1".to_owned()),
-            None,
-            None,
-            None,
-            None,
-        ));
-        let mut data = SessionData::new(
-            storage.weak_ref(),
-            client_url,
-            None,
-            std::sync::Arc::new(crate::FeatureFlags::default()),
-            webhook.clone(),
-        );
-        data.req = Some(req.clone());
-        data.auth_state = SessionAuthState::Authorized;
-        data.binding_version = Some(1);
-        data.webhook_connected_binding_version = Some(1);
-        data.storage_token = Some(StorageToken {
-            token: req.user_token.clone(),
-            client_url: data.client_url.clone(),
-            machine_id,
-            user_id,
-        });
-        data.managed_runtime = storage.bind_managed_runtime_state(user_id, machine_id, None, 0);
-        data.managed_runtime().applied_config_revision = persisted_revision.map(str::to_owned);
-        let session_data = std::sync::Arc::new(RwLock::new(data));
-        let client_url = session_data.read().await.client_url.clone();
-        webhook_validation::apply_success(
-            &std::sync::Arc::downgrade(&session_data),
-            webhook_validation::WebhookValidationInput {
-                storage: storage.clone(),
-                webhook_config: webhook,
-                client_url,
-                applied_config_revision: persisted_revision.map(str::to_owned),
-                applied_config_revision_known: false,
-                failed_instance_ids: Vec::new(),
-                req: req.clone(),
-                machine_id,
-            },
-            webhook_validation::WebhookHeartbeatValidation {
-                config_revision: "current".to_owned(),
-                binding_version: 1,
-            },
-            user_id,
-            0,
-        )
-        .await;
-        let mut round = round_with_revision_state(persisted_revision, false);
-        round.req = req;
-        round.machine_id = machine_id;
-        round.user_id = user_id;
-        round.runtime_config_epoch = session_data
-            .read()
-            .await
-            .managed_runtime()
-            .runtime_config_epoch;
-        (storage, session_data, round)
-    }
-
-    #[tokio::test]
-    async fn takeover_skips_missing_and_outdated_receiver_config() {
-        for revision in [None, Some("outdated")] {
-            let (storage, data, round) = takeover_session(revision).await;
-            let rpc = easytier::proto::rpc::bidirect::BidirectRpcManager::new();
-            let mut client: SessionRpcClient = rpc.rpc_client().scoped_client::<
-                easytier::proto::api::manage::WebClientServiceClientFactory<BaseController>,
-            >(1, 1, String::new());
-
-            assert!(matches!(
-                prepare_reconcile_round(
-                    &std::sync::Arc::downgrade(&data),
-                    &storage.weak_ref().upgrade().unwrap(),
-                    &mut client,
-                    round.req,
-                )
-                .await,
-                RoundStatus::Skip
-            ));
-        }
-    }
-
-    #[tokio::test]
-    async fn takeover_fences_mutations_and_applied_reports_from_unready_rounds() {
-        for revision in [None, Some("outdated")] {
-            let (storage, data, mut round) = takeover_session(revision).await;
-            let weak = std::sync::Arc::downgrade(&data);
-            let mut mutation = RuntimeMutationFence::default();
-            assert!(!begin_managed_runtime_mutation(&weak, &round, &mut mutation).await);
-            assert!(!mutation.started);
-
-            round.should_apply_runtime_revision = true;
-            mark_config_revision_applied_if_current(
-                &weak,
-                &storage.weak_ref().upgrade().unwrap(),
-                &round,
-                &ReconcileOutcome::default(),
-                &mutation,
-            )
-            .await;
-            let data = data.read().await;
-            let runtime = data.managed_runtime();
-            assert!(!runtime.applied_config_revision_known);
-            assert_eq!(runtime.applied_config_revision.as_deref(), revision);
-        }
-    }
-
-    async fn validate_takeover_session(
-        storage: &crate::client_manager::storage::Storage,
-        data: &std::sync::Arc<RwLock<SessionData>>,
-        round: &ReconcileRound,
-        revision: &str,
-        epoch: u64,
-    ) {
-        use super::super::webhook_validation;
-
-        let (webhook_config, client_url) = {
-            let data = data.read().await;
-            (data.webhook_config.clone(), data.client_url.clone())
-        };
-        webhook_validation::apply_success(
-            &std::sync::Arc::downgrade(data),
-            webhook_validation::WebhookValidationInput {
-                storage: storage.clone(),
-                webhook_config,
-                client_url,
-                applied_config_revision: None,
-                applied_config_revision_known: false,
-                failed_instance_ids: Vec::new(),
-                req: round.req.clone(),
-                machine_id: round.machine_id,
-            },
-            webhook_validation::WebhookHeartbeatValidation {
-                config_revision: revision.to_owned(),
-                binding_version: 1,
-            },
-            round.user_id,
-            epoch,
-        )
-        .await;
-    }
-
-    #[tokio::test]
-    async fn takeover_new_publication_fences_late_validation_and_publication_callbacks() {
-        use super::super::{HeartbeatPolicy, Session};
-        use std::sync::Arc;
-
-        let (storage, data, round) = takeover_session(Some("current")).await;
-        let mut session = Session::new(
-            storage.weak_ref(),
-            "tcp://127.0.0.1:1001".parse().unwrap(),
-            None,
-            HeartbeatPolicy::default(),
-            Arc::new(crate::FeatureFlags::default()),
-            data.read().await.webhook_config.clone(),
-            0,
-        );
-        session.data = data.clone();
-        let old_epoch = data.read().await.webhook_validation_change_epoch;
-        storage
-            .db()
-            .set_managed_config_revision((round.user_id, round.machine_id), "newer")
-            .await
-            .unwrap();
-        storage.record_full_managed_config_change(round.user_id, round.machine_id, Some("newer"));
-        session
-            .notify_managed_runtime_state_changed(round.user_id, round.machine_id, Some("newer"))
-            .await;
-
-        validate_takeover_session(&storage, &data, &round, "current", old_epoch).await;
-        session
-            .notify_managed_runtime_state_changed(round.user_id, round.machine_id, Some("current"))
-            .await;
-        let current_epoch = {
-            let data = data.read().await;
-            assert_eq!(data.webhook_config_revision.as_deref(), Some("current"));
-            assert!(!data.runtime_config_is_ready(Some("newer")));
-            assert!(data.webhook_validation_dirty);
-            data.webhook_validation_change_epoch
-        };
-        assert!(
-            !begin_managed_runtime_mutation(
-                &Arc::downgrade(&data),
-                &round,
-                &mut RuntimeMutationFence::default()
-            )
-            .await
-        );
-
-        validate_takeover_session(&storage, &data, &round, "newer", current_epoch).await;
-        assert!(data.read().await.runtime_config_is_ready(Some("newer")));
-
-        // A newer Console revision may be observed after a publication commits
-        // but before its callback runs. SQLite still containing that publication
-        // does not allow the late callback to replace the newer authority.
-        validate_takeover_session(&storage, &data, &round, "newest", current_epoch).await;
-        session
-            .notify_managed_runtime_state_changed(round.user_id, round.machine_id, Some("newer"))
-            .await;
-        let data = data.read().await;
-        assert_eq!(data.webhook_config_revision.as_deref(), Some("newest"));
-        assert!(!data.runtime_config_is_ready(Some("newer")));
-        assert!(data.webhook_validation_dirty);
-    }
-
-    #[tokio::test]
-    async fn unchanged_validation_does_not_replay_an_old_heartbeat() {
-        let (storage, data, round) = takeover_session(Some("current")).await;
-        let mut notifications = data.read().await.heartbeat_waiter();
-        let epoch = data.read().await.webhook_validation_change_epoch;
-        validate_takeover_session(&storage, &data, &round, "current", epoch).await;
-        assert!(matches!(
-            notifications.try_recv(),
-            Err(broadcast::error::TryRecvError::Empty)
-        ));
-
-        let newest_instance = uuid::Uuid::new_v4();
-        {
-            let mut data = data.write().await;
-            data.notifier.send(()).unwrap();
-            data.req.as_mut().unwrap().running_network_instances = vec![newest_instance.into()];
-        }
-        let req = recv_latest_heartbeat(&std::sync::Arc::downgrade(&data), &mut notifications)
-            .await
-            .unwrap();
-        assert_eq!(req.running_network_instances, vec![newest_instance.into()]);
-    }
-
-    #[tokio::test]
-    async fn authoritative_empty_config_remains_ready_but_missing_revision_does_not() {
-        let (storage, data, round) = takeover_session(Some("current")).await;
-        let mut mutation = RuntimeMutationFence::default();
-        assert!(round.local_configs.is_empty());
-        assert!(
-            begin_managed_runtime_mutation(
-                &std::sync::Arc::downgrade(&data),
-                &round,
-                &mut mutation
-            )
-            .await
-        );
-        let epoch = data.read().await.webhook_validation_change_epoch;
-        validate_takeover_session(&storage, &data, &round, "", epoch).await;
-        let mut data = data.write().await;
-        assert!(!data.runtime_config_is_ready(None));
-        assert!(!data.runtime_config_is_ready(Some("current")));
-        data.binding_version = Some(0);
-        assert!(
-            data.runtime_config_is_ready(None),
-            "legacy webhook without managed fields"
-        );
-    }
-
     fn config_with_port_forwards(port_forwards: Vec<PortForwardConfig>) -> NetworkConfig {
         NetworkConfig {
             instance_id: Some("11111111-1111-1111-1111-111111111111".to_string()),

easytier-web/src/client_manager/session/webhook_validation.rs

@@ -393,7 +393,7 @@ pub(super) async fn apply_success(
     validation_change_epoch: u64,
 ) {
     let WebhookHeartbeatValidation {
-        config_revision,
+        config_revision: _,
         binding_version,
     } = validation;
 
@@ -408,7 +408,6 @@ pub(super) async fn apply_success(
         validation_notify,
         runtime_req,
         session_epoch,
-        notify_runtime,
     ) = {
         let mut data = session_data.write().await;
         let Some(runtime_req) = data.req.clone() else {
@@ -437,9 +436,6 @@ pub(super) async fn apply_success(
             return;
         }
 
-        let notify_runtime = !data.auth_state.is_authorized()
-            || data.webhook_config_revision.as_deref() != Some(config_revision.as_str());
-        let runtime_changed = notify_runtime || data.binding_version != Some(binding_version);
         let previous_connected_binding_version = data.webhook_connected_binding_version;
         let client_url = data.client_url.clone();
         let is_new_storage_token = data.storage_token.is_none();
@@ -465,11 +461,6 @@ pub(super) async fn apply_success(
         let storage_token = storage_token.clone();
         data.auth_state = SessionAuthState::Authorized;
         data.binding_version = Some(binding_version);
-        data.webhook_config_revision = Some(config_revision);
-        if runtime_changed {
-            let mut runtime = data.managed_runtime();
-            runtime.runtime_config_epoch = runtime.runtime_config_epoch.wrapping_add(1);
-        }
         if is_new_storage_token {
             tracing::info!(
                 machine_id = %input.machine_id,
@@ -529,7 +520,6 @@ pub(super) async fn apply_success(
             validation_notify,
             runtime_req,
             data.session_epoch,
-            notify_runtime,
         )
     };
 
@@ -551,9 +541,7 @@ pub(super) async fn apply_success(
         .await;
     }
 
-    if notify_runtime {
-        let _ = notifier.send(());
-    }
+    let _ = notifier.send(runtime_req);
 }
 
 async fn wait_webhook_connection_transition(
@@ -575,87 +563,6 @@ async fn wait_webhook_connection_transition(
 mod tests {
     use super::*;
 
-    #[tokio::test]
-    async fn unchanged_validation_does_not_repeat_runtime_notifications() {
-        let storage = Storage::new(crate::db::Db::memory_db().await);
-        let user_id = storage.db().auto_create_user("token").await.unwrap().id;
-        let machine_id = uuid::Uuid::new_v4();
-        let session_data = validation_session(machine_id).await;
-        {
-            let mut data = session_data.write().await;
-            data.storage = storage.weak_ref();
-            data.auth_state = SessionAuthState::Init;
-        }
-        let mut notifications = session_data.read().await.heartbeat_waiter();
-        for attempt in 0..100 {
-            let (input, epoch) = {
-                let data = session_data.read().await;
-                (
-                    WebhookValidationInput {
-                        storage: storage.clone(),
-                        webhook_config: data.webhook_config.clone(),
-                        client_url: data.client_url.clone(),
-                        applied_config_revision: (attempt > 0).then(|| "current".to_owned()),
-                        applied_config_revision_known: attempt > 0,
-                        failed_instance_ids: Vec::new(),
-                        req: data.req.clone().unwrap(),
-                        machine_id,
-                    },
-                    data.webhook_validation_change_epoch,
-                )
-            };
-            apply_success(
-                &Arc::downgrade(&session_data),
-                input,
-                WebhookHeartbeatValidation {
-                    config_revision: "current".to_owned(),
-                    binding_version: if attempt == 99 { 2 } else { 1 },
-                },
-                user_id,
-                epoch,
-            )
-            .await;
-            if attempt == 0 {
-                notifications
-                    .try_recv()
-                    .expect("initial authentication wakes runtime");
-                let data = session_data.read().await;
-                let mut runtime = data.managed_runtime();
-                runtime.applied_config_revision = Some("current".to_owned());
-                runtime.applied_config_revision_known = true;
-            } else {
-                assert!(matches!(
-                    notifications.try_recv(),
-                    Err(tokio::sync::broadcast::error::TryRecvError::Empty)
-                ));
-            }
-        }
-        // A fresh heartbeat still wakes the reconciler after repeated success.
-        let instance_id = uuid::Uuid::new_v4();
-        let mut fresh = session_data.read().await.req.clone().unwrap();
-        fresh.running_network_instances = vec![instance_id.into()];
-        SessionRpcService {
-            data: session_data.clone(),
-            heartbeat_policy: super::super::HeartbeatPolicy::default(),
-        }
-        .handle_heartbeat(fresh)
-        .await
-        .unwrap();
-        notifications
-            .try_recv()
-            .expect("fresh heartbeat wakes runtime");
-        assert_eq!(
-            session_data
-                .read()
-                .await
-                .req
-                .as_ref()
-                .unwrap()
-                .running_network_instances,
-            vec![instance_id.into()]
-        );
-    }
-
     async fn validation_session(machine_id: uuid::Uuid) -> Arc<RwLock<SessionData>> {
         let storage = Storage::new(crate::db::Db::memory_db().await);
         let mut data = SessionData::new(

easytier/src/core.rs

@@ -19,17 +19,14 @@ use crate::{
 };
 use anyhow::Context;
 use cidr::IpCidr;
-use clap::{
-    Arg, ArgMatches, CommandFactory, FromArgMatches, Parser, builder::TypedValueParser as _,
-};
+use clap::{Arg, ArgMatches, CommandFactory, FromArgMatches, Parser};
 use easytier_core::config::{DhcpConfig, EncryptionAlgorithm, normalize_secure_mode_config};
 use guarden::defer;
 use prost_types::field_descriptor_proto::Type;
 use rust_i18n::t;
 use serde_json::{Value, json};
-use std::str::FromStr;
+use std::str::{FromStr, ParseBoolError};
 use std::{
-    fmt::Write as _,
     net::{IpAddr, SocketAddr},
     path::PathBuf,
     process::ExitCode,
@@ -1021,8 +1018,8 @@ fn win_service_event_loop(
 
     let normal_status = ServiceStatus {
         service_type: ServiceType::OWN_PROCESS,
-        current_state: ServiceState::Stopped,
-        controls_accepted: ServiceControlAccept::empty(),
+        current_state: ServiceState::Running,
+        controls_accepted: ServiceControlAccept::STOP,
         exit_code: ServiceExitCode::Win32(0),
         checkpoint: 0,
         wait_hint: Duration::default(),
@@ -1105,9 +1102,9 @@ fn cli_command() -> clap::Command {
         value.parse::<T>().map(|value| json!(value))
     }
 
-    let parser = clap::builder::BoolishValueParser::new();
-    let boolish = parser.map(|v| json!(v));
-    let negated = parser.map(|v| json!(!v));
+    fn negated(value: &str) -> Result<Value, ParseBoolError> {
+        value.parse::<bool>().map(|value| json!(!value))
+    }
 
     Cli::command().args(Flags::flags().iter().map(|field| {
         let name = field.name();
@@ -1118,11 +1115,11 @@ fn cli_command() -> clap::Command {
                 .short('u')
                 .num_args(0..=1)
                 .default_missing_value("true")
-                .value_parser(negated.clone()),
+                .value_parser(negated),
             "enable_ipv6" => arg(name, "disable_ipv6")
                 .num_args(0..=1)
                 .default_missing_value("true")
-                .value_parser(negated.clone()),
+                .value_parser(negated),
             // The command line names the algorithm instead of spelling the field.
             // The values are the schema's own, which the patch reads.
             "data_compress_algo" => {
@@ -1141,7 +1138,7 @@ fn cli_command() -> clap::Command {
                     Type::Bool => arg
                         .num_args(0..=1)
                         .default_missing_value("true")
-                        .value_parser(boolish.clone()),
+                        .value_parser(parsed::<bool>),
                     Type::Uint32 => arg.value_parser(parsed::<u32>),
                     Type::Uint64 => arg.value_parser(parsed::<u64>),
                     // An enum flag takes the name the schema spells, and the patch's
@@ -1212,11 +1209,7 @@ fn win_service_main(arg: Vec<std::ffi::OsString>) {
 
 async fn run_main(cli: Cli) -> anyhow::Result<()> {
     defer!(dump_profile(0););
-    log::init_with_default_console_targets(
-        &cli.logging_options,
-        true,
-        &["CORE", "easytier_core::instance::manager"],
-    )?;
+    log::init(&cli.logging_options, true)?;
 
     let manager = Arc::new(native_cli_instance_manager().with_config_path(cli.config_dir.clone()));
 
@@ -1365,17 +1358,10 @@ async fn run_main(cli: Cli) -> anyhow::Result<()> {
     tokio::select! {
         _ = manager.wait() => {
             let infos = manager.collect_network_infos().await?;
-            let mut errors = String::new();
-            for (instance_id, info) in infos {
-                if let Some(error) = info.error_msg {
-                    if errors.is_empty() {
-                        errors.push_str("some instances stopped with errors:");
-                    }
-                    write!(errors, "\n{instance_id}: {error}").unwrap();
-                }
-            }
-            if !errors.is_empty() {
-                return Err(anyhow::anyhow!(errors));
+            if infos
+                .into_values()
+                .filter_map(|info| info.error_msg).next().is_some() {
+                return Err(anyhow::anyhow!("some instances stopped with errors"));
             }
         }
         _ = tokio::signal::ctrl_c() => {
@@ -1501,7 +1487,7 @@ pub async fn main() -> ExitCode {
     let mut ret_code = 0;
 
     if let Err(error) = run_main(cli).await {
-        log::error!(error = %format_args!("{error:#}"));
+        log::error!(%error);
         ret_code = 1;
     }
 
@@ -1786,44 +1772,6 @@ enabled = true
         let patch = parse_flags(&["easytier", "--disable-ipv6"]).unwrap();
         assert_eq!(patch.enable_ipv6, Some(false));
 
-        // Boolish tests: accepts 1/0, yes/no, on/off, true/false
-        assert_eq!(
-            parse_flags(&["easytier", "--latency-first", "1"])
-                .unwrap()
-                .latency_first,
-            Some(true)
-        );
-        assert_eq!(
-            parse_flags(&["easytier", "--latency-first", "0"])
-                .unwrap()
-                .latency_first,
-            Some(false)
-        );
-        assert_eq!(
-            parse_flags(&["easytier", "--latency-first", "yes"])
-                .unwrap()
-                .latency_first,
-            Some(true)
-        );
-        assert_eq!(
-            parse_flags(&["easytier", "--latency-first", "no"])
-                .unwrap()
-                .latency_first,
-            Some(false)
-        );
-        assert_eq!(
-            parse_flags(&["easytier", "--disable-encryption", "1"])
-                .unwrap()
-                .enable_encryption,
-            Some(false)
-        );
-        assert_eq!(
-            parse_flags(&["easytier", "--disable-encryption", "0"])
-                .unwrap()
-                .enable_encryption,
-            Some(true)
-        );
-
         // Compression tests: case-insensitive, accepts None/none/zstd/Zstd/ZSTD, rejects invalid
         let patch = parse_flags(&["easytier", "--compression", "none"]).unwrap();
         assert_eq!(

easytier/src/socket/fake_tcp/netfilter/linux_bpf.rs

@@ -870,28 +870,20 @@ mod tests {
             let suffix = rand::thread_rng().gen_range(10000..99999);
             let capture_name = format!("elbc{suffix}");
             let sender_name = format!("elbs{suffix}");
-            // Assign MACs at creation so udev does not replace them after readback.
             run_ip(&[
                 "link",
                 "add",
                 &capture_name,
-                "address",
-                "02:00:00:00:00:01",
                 "type",
                 "veth",
                 "peer",
                 "name",
                 &sender_name,
-                "address",
-                "02:00:00:00:00:02",
             ])?;
 
             let setup_result = (|| {
                 run_ip(&["link", "set", "dev", &capture_name, "up"])?;
                 run_ip(&["link", "set", "dev", &sender_name, "up"])?;
-                // A single-device GETLINK synchronizes pending linkwatch work,
-                // activating the first endpoint's TX queue before any test sends.
-                run_ip(&["link", "show", "dev", &capture_name])?;
                 let interfaces = datalink::interfaces();
                 let find_mac = |name: &str| {
                     interfaces

easytier/src/socket/tcp.rs

@@ -303,14 +303,7 @@ pub(crate) async fn bind_tcp_listener(
     let addr = bind_options.local_addr.ok_or_else(|| {
         TunnelError::InvalidAddr("tcp listener requires a local bind address".to_owned())
     })?;
-    if bind_options.bind_device.is_none()
-        && matches!(
-            purpose,
-            TcpListenPurpose::PortLease | TcpListenPurpose::PortForward
-        )
-    {
-        // Port forwards follow the bind IP across interface recreation. An
-        // implicit device binding would retain the deleted interface's index.
+    if bind_options.bind_device.is_none() && purpose == TcpListenPurpose::PortLease {
         bind_options.bind_device = Some(String::new());
     }
     let listener = create_tcp_socket(addr, &bind_options).await?.listen(1024)?;
@@ -370,22 +363,6 @@ mod tests {
 
     use super::*;
 
-    #[cfg(target_os = "linux")]
-    #[rstest::rstest]
-    #[tokio::test]
-    async fn port_forward_listener_honors_device_binding(
-        #[values(None, Some("lo"), Some(""))] device: Option<&str>,
-    ) {
-        let mut options = TcpListenOptions::port_forward("127.0.0.1:0".parse().unwrap());
-        options.bind.bind_device = device.map(str::to_owned);
-        let listener = bind_tcp_listener(options).await.unwrap();
-        let bound = SockRef::from(&listener.listener).device().unwrap();
-        assert_eq!(
-            bound.as_deref(),
-            device.filter(|name| !name.is_empty()).map(str::as_bytes)
-        );
-    }
-
     #[tokio::test]
     async fn runtime_tcp_owned_split_preserves_full_duplex_and_shutdown() {
         let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();

easytier/src/socket/udp.rs

@@ -146,17 +146,27 @@ impl RuntimeUdpSocketFactory {
             return BindDev::from(bind_device.as_str());
         }
 
-        // Port forwards bind only their local address by default, so replacing
-        // its interface does not strand the socket on the old interface index.
         if matches!(
             options.purpose,
-            UdpSocketPurpose::DirectConnect | UdpSocketPurpose::PortBoundListener
+            UdpSocketPurpose::DirectConnect
+                | UdpSocketPurpose::PortBoundListener
+                | UdpSocketPurpose::PortForward
         ) {
             return BindDev::Auto;
         }
 
         BindDev::Disabled
     }
+
+    fn reuse_addr_for(&self, options: &UdpBindOptions) -> bool {
+        options.reuse_addr
+            || (matches!(
+                options.purpose,
+                UdpSocketPurpose::PortBoundListener
+                    | UdpSocketPurpose::ProxyNat
+                    | UdpSocketPurpose::PortForward
+            ) && !cfg!(target_os = "windows"))
+    }
 }
 
 /// Creates a host-owned UDP socket, including any required protection, before
@@ -171,9 +181,7 @@ pub(crate) async fn create_udp_socket(options: &UdpBindOptions) -> anyhow::Resul
         .dev(factory.bind_device_for(options))
         .net_ns(NetNS::from_socket_context(&options.context))
         .only_v6(options.only_v6)
-        // Independent UDP receivers must not share a bound endpoint, including
-        // kernel-assigned ports. Only an explicit request may enable reuse.
-        .reuse_addr(options.reuse_addr)
+        .reuse_addr(factory.reuse_addr_for(options))
         .reuse_port(options.reuse_port)
         .maybe_socket_mark(options.context.socket_mark)
         .need_protect(options.need_protect)
@@ -208,78 +216,6 @@ mod tests {
 
     use super::*;
 
-    #[cfg(target_os = "linux")]
-    #[tokio::test]
-    async fn runtime_udp_bind_rejects_occupied_port() {
-        for local_addr in ["0.0.0.0:0", "[::]:0"] {
-            let local_addr = local_addr.parse().unwrap();
-            for bind in [
-                UdpBindOptions::port_bound_listener(local_addr),
-                UdpBindOptions::proxy_nat().with_local_addr(Some(local_addr)),
-                UdpBindOptions::port_forward(local_addr),
-            ] {
-                let bind = bind.with_only_v6(true);
-                let listener = create_udp_socket(&bind).await.unwrap();
-                let occupied = bind.with_local_addr(Some(listener.local_addr().unwrap()));
-
-                for reuse_addr in [false, true] {
-                    let error = create_udp_socket(&occupied.clone().with_reuse_addr(reuse_addr))
-                        .await
-                        .expect_err("another socket must not share a bound receive endpoint");
-                    assert!(matches!(
-                        error.downcast_ref::<easytier_core::tunnel::TunnelError>(),
-                        Some(easytier_core::tunnel::TunnelError::IOError(error))
-                            if error.kind() == std::io::ErrorKind::AddrInUse
-                    ));
-                }
-
-                drop(listener);
-                create_udp_socket(&occupied).await.unwrap();
-            }
-        }
-    }
-
-    #[tokio::test]
-    async fn runtime_udp_listeners_keep_ephemeral_ports_and_packets_separate() {
-        for (bind_addr, loopback) in [
-            ("0.0.0.0:0", "127.0.0.1".parse::<IpAddr>().unwrap()),
-            ("[::]:0", "::1".parse::<IpAddr>().unwrap()),
-        ] {
-            let bind_addr = bind_addr.parse().unwrap();
-            let binds = [
-                UdpBindOptions::port_bound_listener(bind_addr),
-                UdpBindOptions::proxy_nat().with_local_addr(Some(bind_addr)),
-                UdpBindOptions::port_forward(bind_addr),
-            ];
-            let mut listeners = Vec::new();
-            let mut ports = std::collections::HashSet::new();
-            for index in 0..128 {
-                let bind = binds[index % binds.len()].clone().with_only_v6(true);
-                let listener = create_udp_socket(&bind).await.unwrap();
-                assert!(ports.insert(listener.local_addr().unwrap().port()));
-                listeners.push(listener);
-            }
-
-            let sender = UdpSocket::bind(SocketAddr::new(loopback, 0)).await.unwrap();
-            for (index, listener) in listeners.iter().enumerate() {
-                let destination = SocketAddr::new(loopback, listener.local_addr().unwrap().port());
-                sender.send_to(&[index as u8], destination).await.unwrap();
-            }
-            for (index, listener) in listeners.iter().enumerate() {
-                let mut packet = [0; 16];
-                let (len, source) = tokio::time::timeout(
-                    std::time::Duration::from_secs(2),
-                    listener.recv_from(&mut packet),
-                )
-                .await
-                .expect("listener did not receive its packet")
-                .unwrap();
-                assert_eq!(&packet[..len], &[index as u8]);
-                assert_eq!(source, sender.local_addr().unwrap());
-            }
-        }
-    }
-
     #[cfg(any(target_os = "linux", target_os = "android"))]
     #[tokio::test]
     async fn runtime_udp_socket_reports_ipv4_destination_ip() {
@@ -424,7 +360,7 @@ mod tests {
     }
 
     #[test]
-    fn factory_interprets_bind_device_by_purpose() {
+    fn factory_interprets_bind_defaults_by_purpose() {
         let listener_addr = SocketAddr::from(([0, 0, 0, 0], 11010));
         let factory = RuntimeUdpSocketFactory::new();
 
@@ -438,7 +374,7 @@ mod tests {
         ));
         assert!(matches!(
             factory.bind_device_for(&UdpBindOptions::port_forward(listener_addr)),
-            BindDev::Disabled
+            BindDev::Auto
         ));
         assert!(matches!(
             factory.bind_device_for(&UdpBindOptions::port_lease(listener_addr)),
@@ -448,18 +384,28 @@ mod tests {
             factory.bind_device_for(&UdpBindOptions::hole_punch_control()),
             BindDev::Disabled
         ));
+        assert_eq!(
+            factory.reuse_addr_for(&UdpBindOptions::port_bound_listener(listener_addr)),
+            !cfg!(target_os = "windows")
+        );
+        assert!(!factory.reuse_addr_for(&UdpBindOptions::hole_punch_control()));
+        assert_eq!(
+            factory.reuse_addr_for(&UdpBindOptions::proxy_nat()),
+            !cfg!(target_os = "windows")
+        );
+        assert_eq!(
+            factory.reuse_addr_for(&UdpBindOptions::port_forward(listener_addr)),
+            !cfg!(target_os = "windows")
+        );
+        assert!(!factory.reuse_addr_for(&UdpBindOptions::port_lease(listener_addr)));
     }
 
-    #[rstest::rstest]
-    fn factory_applies_listener_bind_device_option(#[values(false, true)] port_forward: bool) {
+    #[test]
+    fn factory_applies_listener_bind_device_option() {
         let listener_addr = SocketAddr::from(([0, 0, 0, 0], 11010));
         let factory = RuntimeUdpSocketFactory::new();
-        let options = if port_forward {
-            UdpBindOptions::port_forward(listener_addr)
-        } else {
-            UdpBindOptions::port_bound_listener(listener_addr)
-        }
-        .with_bind_device(Some("eth0".to_owned()));
+        let options = UdpBindOptions::port_bound_listener(listener_addr)
+            .with_bind_device(Some("eth0".to_owned()));
 
         match factory.bind_device_for(&options) {
             BindDev::Custom(dev) => assert_eq!(dev, "eth0"),

easytier/src/tests/shared_virtual_nic.rs

@@ -554,24 +554,18 @@ fn shared_tun_claim_cleanup_outlives_member_runtime() {
 #[case::static_ipv4(false)]
 #[case::with_dhcpv6(true)]
 #[tokio::test]
+#[serial_test::serial]
 async fn same_namespace_members_share_tun_across_independent_networks(#[case] dhcpv6: bool) {
     let dev_name = test_dev_name();
-    let mut network = TestNetwork::new();
-    // Nextest runs each case in its own process, so their shared TUNs must
-    // also have separate namespaces to avoid competing routes for the same IPs.
-    network.create_empty_namespace("shared");
-    network.create_namespace("first_peer", "10.231.1.2/24", "fd31::2/64");
-    network.create_namespace("second_peer", "10.231.2.2/24", "fd32::2/64");
-    let shared_ns = network.namespace("shared");
-    let first_peer_ns = network.namespace("first_peer");
-    let second_peer_ns = network.namespace("second_peer");
+    let first_peer_ns = TestNetnsGuard::new(short_name("sva"), "10.231.1.2/24", "fd31::2/64");
+    let second_peer_ns = TestNetnsGuard::new(short_name("svb"), "10.231.2.2/24", "fd32::2/64");
     let runtime = SharedTestRuntime::new();
 
     let first_config = test_config(
         "shared_tun_first",
         "shared_tun_network_a",
         "shared_tun_secret_a",
-        Some(shared_ns),
+        None,
         Some(&dev_name),
         "10.144.250.1/24",
     );
@@ -585,7 +579,7 @@ async fn same_namespace_members_share_tun_across_independent_networks(#[case] dh
         "shared_tun_second",
         "shared_tun_network_b",
         "shared_tun_secret_b",
-        Some(shared_ns),
+        None,
         Some(&dev_name),
         "10.144.251.1/24",
     );
@@ -599,7 +593,7 @@ async fn same_namespace_members_share_tun_across_independent_networks(#[case] dh
         "shared_tun_first_peer",
         "shared_tun_network_a",
         "shared_tun_secret_a",
-        Some(first_peer_ns),
+        Some(&first_peer_ns.name),
         None,
         "10.144.250.2/24",
     );
@@ -611,7 +605,7 @@ async fn same_namespace_members_share_tun_across_independent_networks(#[case] dh
         "shared_tun_second_peer",
         "shared_tun_network_b",
         "shared_tun_secret_b",
-        Some(second_peer_ns),
+        Some(&second_peer_ns.name),
         None,
         "10.144.251.2/24",
     );
@@ -654,18 +648,21 @@ async fn same_namespace_members_share_tun_across_independent_networks(#[case] dh
     )
     .await;
     wait_for_condition(
-        || async { ping_test(first_peer_ns, "10.144.250.1", None).await },
+        || async { ping_test(&first_peer_ns.name, "10.144.250.1", None).await },
         WAIT,
     )
     .await;
     wait_for_condition(
-        || async { ping_test(second_peer_ns, "10.144.251.1", None).await },
+        || async { ping_test(&second_peer_ns.name, "10.144.251.1", None).await },
         WAIT,
     )
     .await;
 
     if dhcpv6 {
-        for (instance, ns) in [(&first, first_peer_ns), (&second, second_peer_ns)] {
+        for (instance, ns) in [
+            (&first, &first_peer_ns.name),
+            (&second, &second_peer_ns.name),
+        ] {
             wait_for_condition(
                 || async { instance.get_global_ctx().get_ipv6().is_some() },
                 WAIT,
@@ -681,10 +678,10 @@ async fn same_namespace_members_share_tun_across_independent_networks(#[case] dh
         }
         // Removing one DHCP member must leave the other member's addresses
         // and the physical TUN intact.
-        let original_index = tun_ifindex(shared_ns, &dev_name);
+        let original_index = tun_ifindex(ROOT_NETNS_NAME, &dev_name);
         assert!(original_index.is_some());
         drop_insts(vec![first]).await;
-        assert_eq!(tun_ifindex(shared_ns, &dev_name), original_index);
+        assert_eq!(tun_ifindex(ROOT_NETNS_NAME, &dev_name), original_index);
         let address = second
             .get_global_ctx()
             .get_ipv6()
@@ -692,7 +689,7 @@ async fn same_namespace_members_share_tun_across_independent_networks(#[case] dh
             .address()
             .to_string();
         wait_for_condition(
-            || async { ping_test(second_peer_ns, &address, None).await },
+            || async { ping_test(&second_peer_ns.name, &address, None).await },
             WAIT,
         )
         .await;

easytier/src/tests/three_node.rs

@@ -2075,13 +2075,8 @@ pub async fn wireguard_vpn_portal(#[values(true, false)] test_v6: bool) {
                 config
                     .add_proxy_cidr("198.51.100.0/24".parse().unwrap(), None)
                     .unwrap();
-                let mut flags = config.get_flags();
-                flags.enable_exit_node = true;
-                flags.use_smoltcp = true;
-                config.set_flags(flags);
             }
             if config.get_inst_name() == "inst3" {
-                config.set_exit_nodes(vec!["10.144.144.1".parse().unwrap()]);
                 config.set_vpn_portal_config(VpnPortalConfig {
                     enabled: None,
                     wireguard_listen: "0.0.0.0:22121".parse().unwrap(),
@@ -2142,7 +2137,7 @@ pub async fn wireguard_vpn_portal(#[values(true, false)] test_v6: bool) {
         dst_socket_addr,
         Key::try_from(server_public.as_slice()).unwrap(),
         Key::try_from(client_private.as_slice()).unwrap(),
-        vec!["10.144.144.0/24".to_string(), "10.1.1.1/32".to_string()],
+        vec!["10.144.144.0/24".to_string()],
         "10.144.144.4".to_owned(),
     )
     .unwrap();
@@ -2166,14 +2161,6 @@ pub async fn wireguard_vpn_portal(#[values(true, false)] test_v6: bool) {
     )
     .await;
 
-    // This address is outside the mesh and advertised proxy CIDRs, on a
-    // separate bridge reachable only through the configured exit node.
-    wait_for_condition(
-        || async { ping_test(network.namespace("net_d"), "10.1.1.1", None).await },
-        Duration::from_secs(5),
-    )
-    .await;
-
     drop_insts(insts).await;
 }
 
@@ -2937,84 +2924,6 @@ pub async fn manual_reconnector(#[values(true, false)] is_foreign: bool) {
     drop_insts(vec![center_inst, inst1, inst2]).await;
 }
 
-#[rstest::rstest]
-#[tokio::test]
-#[serial_test::serial]
-async fn port_forward_survives_interface_recreation(#[values("tcp", "udp")] proto: &str) {
-    let network = prepare_linux_namespaces();
-    let namespace = network.namespace("net_a");
-    let create_tun = || {
-        run_ip_in_ns(
-            namespace,
-            &["tuntap", "add", "dev", "pf_tun", "mode", "tun"],
-        );
-        run_ip_in_ns(namespace, &["addr", "add", "192.0.2.1/24", "dev", "pf_tun"]);
-        run_ip_in_ns(namespace, &["link", "set", "pf_tun", "up"]);
-        run_ip_in_ns_output(namespace, &["-o", "link", "show", "pf_tun"])
-            .split(':')
-            .next()
-            .unwrap()
-            .to_owned()
-    };
-    let original_index = create_tun();
-    let insts = init_three_node_ex(
-        &network,
-        "udp",
-        |cfg| {
-            if cfg.get_inst_name() == "inst1" {
-                cfg.set_port_forwards(vec![PortForwardConfig {
-                    bind_addr: "192.0.2.1:23456".parse().unwrap(),
-                    dst_addr: "10.144.144.3:23456".parse().unwrap(),
-                    proto: proto.to_owned(),
-                }]);
-            }
-            cfg
-        },
-        false,
-    )
-    .await;
-
-    for recreated in [false, true] {
-        if recreated {
-            run_ip_in_ns(namespace, &["link", "del", "pf_tun"]);
-            assert_ne!(create_tun(), original_index);
-        }
-        // Keep the forwarding instance and its listener alive across deletion.
-        // Reusing the interface name and IP must not leave it bound to the old
-        // ifindex. Check an application round trip, including the return path.
-        let listener_url = format!("{proto}://0.0.0.0:23456").parse().unwrap();
-        let connector_url = format!("{proto}://192.0.2.1:23456").parse().unwrap();
-        let server_ns = NetNS::new(Some(network.namespace("net_c").into()));
-        let client_ns = NetNS::new(Some(namespace.into()));
-        let payload = b"port-forward interface recreation".to_vec();
-        let timeout = Duration::from_secs(5);
-        let result = if proto == "tcp" {
-            _tunnel_pingpong_netns_with_timeout(
-                core_tcp_listener(listener_url),
-                core_tcp_dialer(connector_url),
-                server_ns,
-                client_ns,
-                payload,
-                timeout,
-            )
-            .await
-        } else {
-            _tunnel_pingpong_netns_with_timeout(
-                core_udp_listener(listener_url),
-                core_udp_dialer(connector_url),
-                server_ns,
-                client_ns,
-                payload,
-                timeout,
-            )
-            .await
-        };
-        assert!(result.is_ok(), "{proto}, recreated={recreated}: {result:?}");
-    }
-
-    drop_insts(insts).await;
-}
-
 #[rstest::rstest]
 #[tokio::test]
 #[serial_test::serial]
@@ -5491,45 +5400,14 @@ pub async fn dhcpv6_three_node_connectivity(#[case] dhcp_ipv4: bool, #[case] sta
     for address in &addresses {
         assert_eq!(address.network(), "fd42:1234::/64".parse().unwrap());
     }
-    // DHCP can temporarily assign the same IPv4 address to multiple peers.
-    // Wait for conflict resolution and route propagation in both directions
-    // before opening connections; IPv6 readiness alone does not imply this.
     wait_for_condition(
         || async {
-            let ipv4_addresses = insts
+            insts[0]
+                .get_core_instance()
+                .route_snapshots()
+                .await
                 .iter()
-                .map(|inst| inst.get_global_ctx().get_ipv4())
-                .collect::<Vec<_>>();
-            if (dhcp_ipv4 || static_ipv4)
-                && ipv4_addresses
-                    .iter()
-                    .flatten()
-                    .map(|address| address.address())
-                    .collect::<std::collections::HashSet<_>>()
-                    .len()
-                    != insts.len()
-            {
-                return false;
-            }
-            for inst in &insts {
-                let routes = inst.get_core_instance().route_snapshots().await;
-                if routes.len() != insts.len() - 1 {
-                    return false;
-                }
-                for (index, peer) in insts.iter().enumerate() {
-                    if peer.peer_id() == inst.peer_id() {
-                        continue;
-                    }
-                    if !routes.iter().any(|route| {
-                        route.peer_id == peer.peer_id()
-                            && route.ipv6_addr == Some(addresses[index].into())
-                            && route.ipv4_addr == ipv4_addresses[index].map(Into::into)
-                    }) {
-                        return false;
-                    }
-                }
-            }
-            true
+                .all(|route| route.ipv6_addr.is_some())
         },
         Duration::from_secs(15),
     )

easytier/src/tunnel/wireguard.rs

@@ -843,32 +843,13 @@ pub mod tests {
     }
 
     #[tokio::test]
-    async fn bind_ipv4_and_ipv6_same_port() {
-        let mut listener = WgTunnelListener::new("wg://[::]:0".parse().unwrap(), test_wg_config());
-        listener.listen().await.unwrap();
-        let port = listener.local_url().port().unwrap();
-
-        let mut listener2 = WgTunnelListener::new(
-            format!("wg://0.0.0.0:{port}").parse().unwrap(),
-            test_wg_config(),
-        );
-        listener2.listen().await.unwrap();
-        assert_eq!(listener2.local_url().port(), Some(port));
-    }
-
-    #[tokio::test]
-    async fn bind_same_port_is_rejected() {
+    async fn bind_same_port() {
         let server_cfg = test_wg_config();
-        let mut listener = WgTunnelListener::new("wg://[::1]:0".parse().unwrap(), server_cfg);
+        let mut listener = WgTunnelListener::new("wg://[::1]:31015".parse().unwrap(), server_cfg);
+        let server_cfg = test_wg_config();
+        let mut listener2 = WgTunnelListener::new("wg://[::1]:31015".parse().unwrap(), server_cfg);
         listener.listen().await.unwrap();
-
-        let mut listener2 = WgTunnelListener::new(listener.local_url(), test_wg_config());
-        let error = listener2.listen().await.unwrap_err();
-        assert!(error.chain().any(|cause| {
-            cause
-                .downcast_ref::<std::io::Error>()
-                .is_some_and(|error| error.kind() == std::io::ErrorKind::AddrInUse)
-        }));
+        listener2.listen().await.unwrap();
     }
 
     #[tokio::test]