tcp和udp分开

README.md

@@ -284,3 +284,39 @@ sudo /home/mynetspeeder/scripts/start-transparent.sh --capture-uid 996 /home/myn
 ```bash
 sudo /home/mynetspeeder/scripts/start-transparent.sh --enable-udp --capture-uid 996 /home/mynetspeeder/config.json
 ```
+
+当前版本新增更激进的 UDP 冗余参数：
+
+```json
+{
+  "udp_redundancy": 1,
+  "udp_direct_redundancy": 2,
+  "udp_always_broadcast": true,
+  "udp_copy_interval_ms": 8
+}
+```
+
+说明：
+
+- `udp_redundancy`：每个 UDP 包额外重复发送的次数
+- `udp_direct_redundancy`：本地 direct UDP 并发副本数
+- `udp_always_broadcast`：即使已有 winner，后续包仍持续并发发往所有可用路径
+- `udp_copy_interval_ms`：多副本之间的间隔，单位毫秒
+
+默认策略更偏向抗丢包和稳态可用，而不是节省流量。
+
+如果你希望同时启用本机显式 SOCKS5 出站入口，只需要在 `config.json` 增加端口：
+
+```json
+{
+  "socks_host": "127.0.0.1",
+  "socks_port": 19180
+}
+```
+
+说明：
+
+- `socks_port` 为 `0` 或不填：不启动
+- `socks_port` 大于 `0`：`start-transparent.sh` 会自动一并启动
+- 适合让 `sing-box` 把指定 UDP/QUIC 流量显式交给 `mynetspeeder`
+- 不需要额外手动执行单独脚本

cli.py

@@ -11,6 +11,7 @@ from . import __version__
 from .config import Config
 from .relay_server import RelayServer
 from .relay_client import RelayManager
+from .socks_edge import SocksEdge
 from .transparent_edge import TransparentEdge
 
 
@@ -45,6 +46,12 @@ def build_parser() -> argparse.ArgumentParser:
     edge.add_argument("--kernel", choices=("auto", "20", "24"), default="auto")
     edge.set_defaults(handler=handle_edge)
 
+    socks = sub.add_parser("socks", help="在当前主 VPS 上启动显式 SOCKS5 出站加速")
+    socks.add_argument("--listen-host", default="127.0.0.1")
+    socks.add_argument("--listen-port", type=int, default=19180)
+    socks.add_argument("--config", required=True)
+    socks.set_defaults(handler=handle_socks)
+
     probe = sub.add_parser("probe", help="查看子节点探测与在线状态")
     probe.add_argument("--config", required=True)
     probe.add_argument("--once", action="store_true")
@@ -68,6 +75,11 @@ def handle_edge(args: argparse.Namespace) -> int:
     return 0
 
 
+def handle_socks(args: argparse.Namespace) -> int:
+    asyncio.run(SocksEdge(args.listen_host, args.listen_port, Config.load(args.config)).start())
+    return 0
+
+
 def handle_probe(args: argparse.Namespace) -> int:
     async def run_probe() -> None:
         manager = RelayManager(Config.load(args.config))

config.json

@@ -9,9 +9,9 @@
   "probe_interval": 3,
   "relay_reconnect_delay": 1,
   "relay_reconnect_max_delay": 10,
+  "socks_host": "127.0.0.1",
+  "socks_port": 19180,
   "relays": [
-    {"name": "hk1", "host": "23.95.134.159", "port": 9009, "token": "130", "weight": 100},
-    {"name": "hk2", "host": "23.238.9.140", "port": 9009, "token": "130", "weight": 100},
-    {"name": "hk3", "host": "23.94.107.3", "port": 9009, "token": "130", "weight": 100}
+    {"name": "hk1", "host": "23.95.134.159", "port": 9009, "token": "130", "weight": 100}
   ] 
 }

config.py

@@ -40,6 +40,12 @@ class Config:
     direct_redundancy_v4: int | None = None
     direct_redundancy_v6: int | None = None
     direct_max_redundancy: int = 3
+    udp_redundancy: int = 1
+    udp_direct_redundancy: int = 2
+    udp_always_broadcast: bool = True
+    udp_copy_interval_ms: int = 8
+    socks_host: str = "127.0.0.1"
+    socks_port: int = 0
 
     @classmethod
     def load(cls, path: str) -> "Config":
@@ -66,4 +72,10 @@ class Config:
             direct_redundancy_v4=raw.get("direct_redundancy_v4"),
             direct_redundancy_v6=raw.get("direct_redundancy_v6"),
             direct_max_redundancy=max(1, raw.get("direct_max_redundancy", 3)),
+            udp_redundancy=max(0, raw.get("udp_redundancy", 1)),
+            udp_direct_redundancy=max(1, raw.get("udp_direct_redundancy", 2)),
+            udp_always_broadcast=raw.get("udp_always_broadcast", True),
+            udp_copy_interval_ms=max(0, raw.get("udp_copy_interval_ms", 8)),
+            socks_host=raw.get("socks_host", "127.0.0.1"),
+            socks_port=max(0, raw.get("socks_port", 0)),
         )

scripts/start-transparent.sh

@@ -19,6 +19,7 @@ ENABLE_UDP=0
 KERNEL_MODE="${MYNETSPEEDER_KERNEL_MODE:-auto}"
 CONFIG_PATH="/home/mynetspeeder/config.json"
 CAPTURE_UID="${MYNETSPEEDER_CAPTURE_UID:-}"
+UDP_CAPTURE_EFFECTIVE=0
 
 while [[ $# -gt 0 ]]; do
   case "$1" in
@@ -45,7 +46,9 @@ LISTEN_HOST="${MYNETSPEEDER_LISTEN_HOST:-127.0.0.1}"
 LISTEN_PORT="${MYNETSPEEDER_LISTEN_PORT:-19080}"
 RUNTIME_USER="${MYNETSPEEDER_USER:-mynetspeeder}"
 PID_FILE="/var/run/mynetspeeder-edge.pid"
+SOCKS_PID_FILE="/var/run/mynetspeeder-socks.pid"
 LOG_FILE="/var/log/mynetspeeder-edge.log"
+SOCKS_LOG_FILE="/var/log/mynetspeeder-socks.log"
 LOG_MAX_MB="${MYNETSPEEDER_LOG_MAX_MB:-50}"
 LOG_BACKUPS="${MYNETSPEEDER_LOG_BACKUPS:-3}"
 CHAIN4="MYNETSPEEDER"
@@ -86,6 +89,8 @@ id -u "$RUNTIME_USER" >/dev/null 2>&1 || useradd --system --no-create-home --she
 mkdir -p /var/log
 touch "$LOG_FILE"
 chown "$RUNTIME_USER":"$RUNTIME_USER" "$LOG_FILE"
+touch "$SOCKS_LOG_FILE"
+chown "$RUNTIME_USER":"$RUNTIME_USER" "$SOCKS_LOG_FILE"
 if ! [[ "$LOG_MAX_MB" =~ ^[0-9]+$ ]] || ! [[ "$LOG_BACKUPS" =~ ^[0-9]+$ ]]; then echo "log limits must be numeric"; exit 1; fi
 LOG_MAX_BYTES=$((LOG_MAX_MB * 1024 * 1024))
 
@@ -107,6 +112,7 @@ ensure_rule() {
 }
 
 add_exclusions_v4() {
+  iptables -t nat -A "$CHAIN4" -m addrtype --dst-type LOCAL -j RETURN
   for cidr in $SELF_EXCLUDE_V4; do
     iptables -t nat -A "$CHAIN4" -d "$cidr" -j RETURN
   done
@@ -126,6 +132,7 @@ PY
 }
 
 add_exclusions_v6() {
+  ip6tables -t nat -A "$CHAIN6" -m addrtype --dst-type LOCAL -j RETURN
   for cidr in $SELF_EXCLUDE_V6; do
     ip6tables -t nat -A "$CHAIN6" -d "$cidr" -j RETURN
   done
@@ -153,8 +160,37 @@ add_udp_exclusions_v6() {
 }
 
 pkill -f 'python3 -m mynetspeeder edge' || true
+pkill -f 'python3 -m mynetspeeder socks' || true
+
+SOCKS_HOST=$(python3 - <<'PY' "$CONFIG_PATH"
+import json, sys
+cfg = json.load(open(sys.argv[1]))
+print(cfg.get('socks_host', '127.0.0.1'))
+PY
+)
+SOCKS_PORT=$(python3 - <<'PY' "$CONFIG_PATH"
+import json, sys
+cfg = json.load(open(sys.argv[1]))
+print(int(cfg.get('socks_port', 0) or 0))
+PY
+)
+
+if [[ "$SOCKS_PORT" -gt 0 ]]; then
+  runuser -u "$RUNTIME_USER" -- bash -lc "export PYTHONUNBUFFERED=1; export PYTHONPATH=/home; cd /home && exec nohup python3 -m mynetspeeder socks --listen-host ${SOCKS_HOST} --listen-port ${SOCKS_PORT} --config ${CONFIG_PATH} 2>&1 | python3 /home/mynetspeeder/scripts/rotate-log.py ${SOCKS_LOG_FILE} ${LOG_MAX_BYTES} ${LOG_BACKUPS}" &
+  SOCKS_PID=$!
+  echo "$SOCKS_PID" > "$SOCKS_PID_FILE"
+  sleep 1
+  ss -lntp | grep -qE "${SOCKS_HOST//./\\.}:${SOCKS_PORT}( |$)" || { echo "socks failed to listen"; tail -n 50 "$SOCKS_LOG_FILE" || true; exit 1; }
+fi
+
+if [[ "$ENABLE_UDP" == "1" && "$SOCKS_PORT" -gt 0 ]]; then
+  echo "udp transparent capture disabled: socks5 is enabled, UDP will use socks only"
+else
+  UDP_CAPTURE_EFFECTIVE="$ENABLE_UDP"
+fi
+
 EDGE_UDP_FLAG=""
-if [[ "$ENABLE_UDP" == "1" ]]; then
+if [[ "$UDP_CAPTURE_EFFECTIVE" == "1" ]]; then
   EDGE_UDP_FLAG="--enable-udp"
 fi
 runuser -u "$RUNTIME_USER" -- bash -lc "export PYTHONUNBUFFERED=1; export PYTHONPATH=/home; cd /home && exec nohup python3 -m mynetspeeder edge --listen-host ${LISTEN_HOST} --listen-port ${LISTEN_PORT} --kernel ${KERNEL_MODE} --config ${CONFIG_PATH} ${EDGE_UDP_FLAG} 2>&1 | python3 /home/mynetspeeder/scripts/rotate-log.py ${LOG_FILE} ${LOG_MAX_BYTES} ${LOG_BACKUPS}" &
@@ -172,7 +208,7 @@ else
   iptables -t nat -A "$CHAIN4" -p tcp -j REDIRECT --to-ports "$LISTEN_PORT"
 fi
 ensure_rule iptables nat OUTPUT -p tcp -j "$CHAIN4"
-if [[ "$ENABLE_UDP" == "1" ]]; then
+if [[ "$UDP_CAPTURE_EFFECTIVE" == "1" ]]; then
   add_udp_exclusions_v4
   if [[ -n "$CAPTURE_UID" ]]; then
     iptables -t nat -A "$CHAIN4" -p udp -m owner --uid-owner "$CAPTURE_UID" -j REDIRECT --to-ports "$LISTEN_PORT"
@@ -197,7 +233,7 @@ if command -v ip6tables >/dev/null 2>&1; then
       ip6tables -t nat -A "$CHAIN6" -p tcp -j REDIRECT --to-ports "$LISTEN_PORT"
     fi
     ensure_rule ip6tables nat OUTPUT -p tcp -j "$CHAIN6"
-    if [[ "$ENABLE_UDP" == "1" ]]; then
+    if [[ "$UDP_CAPTURE_EFFECTIVE" == "1" ]]; then
       add_udp_exclusions_v6
       if [[ -n "$CAPTURE_UID" ]]; then
         ip6tables -t nat -A "$CHAIN6" -p udp -m owner --uid-owner "$CAPTURE_UID" -j REDIRECT --to-ports "$LISTEN_PORT"
@@ -217,12 +253,12 @@ if [[ "$IP6_ENABLED" == "1" && "$IP6_NAT_SUPPORTED" == "1" ]]; then
   RULES_V6=$(ip6tables -t nat -S "$CHAIN6" 2>/dev/null | wc -l)
 fi
 iptables -t nat -C OUTPUT -p tcp -j "$CHAIN4" >/dev/null 2>&1 || { echo "self-check failed: ipv4 tcp output hook missing"; exit 1; }
-if [[ "$ENABLE_UDP" == "1" ]]; then
+if [[ "$UDP_CAPTURE_EFFECTIVE" == "1" ]]; then
   iptables -t nat -C OUTPUT -p udp -j "$CHAIN4" >/dev/null 2>&1 || { echo "self-check failed: ipv4 udp output hook missing"; exit 1; }
 fi
 if [[ "$IP6_ENABLED" == "1" && "$IP6_NAT_SUPPORTED" == "1" ]]; then
   ip6tables -t nat -C OUTPUT -p tcp -j "$CHAIN6" >/dev/null 2>&1 || { echo "self-check failed: ipv6 tcp output hook missing"; exit 1; }
-  if [[ "$ENABLE_UDP" == "1" ]]; then
+  if [[ "$UDP_CAPTURE_EFFECTIVE" == "1" ]]; then
     ip6tables -t nat -C OUTPUT -p udp -j "$CHAIN6" >/dev/null 2>&1 || { echo "self-check failed: ipv6 udp output hook missing"; exit 1; }
   fi
 fi
@@ -240,7 +276,14 @@ if [[ ${#SSH_PORT_ARRAY[@]} -gt 0 ]]; then
 else
   echo "ssh exempt ports: none"
 fi
-echo "udp capture: $ENABLE_UDP"
+echo "udp capture requested: $ENABLE_UDP"
+echo "udp capture effective: $UDP_CAPTURE_EFFECTIVE"
+if [[ "$SOCKS_PORT" -gt 0 ]]; then
+  echo "socks5: ${SOCKS_HOST}:${SOCKS_PORT}"
+  echo "socks log: $SOCKS_LOG_FILE"
+else
+  echo "socks5: disabled"
+fi
 echo "log file: $LOG_FILE"
 echo "log max: ${LOG_MAX_MB}MB x ${LOG_BACKUPS}"
 echo "ipv4 chain rules: $RULES_V4"

scripts/stop-transparent.sh

@@ -4,6 +4,7 @@ set -euo pipefail
 CHAIN4="MYNETSPEEDER"
 CHAIN6="MYNETSPEEDER6"
 PID_FILE="/var/run/mynetspeeder-edge.pid"
+SOCKS_PID_FILE="/var/run/mynetspeeder-socks.pid"
 
 if [[ $EUID -ne 0 ]]; then
   echo "need root"
@@ -14,7 +15,12 @@ if [[ -f "$PID_FILE" ]]; then
   kill "$(cat "$PID_FILE")" 2>/dev/null || true
   rm -f "$PID_FILE"
 fi
+if [[ -f "$SOCKS_PID_FILE" ]]; then
+  kill "$(cat "$SOCKS_PID_FILE")" 2>/dev/null || true
+  rm -f "$SOCKS_PID_FILE"
+fi
 pkill -f 'python3 -m mynetspeeder edge' || true
+pkill -f 'python3 -m mynetspeeder socks' || true
 
 if iptables -t nat -S >/dev/null 2>&1; then
   iptables -t nat -D OUTPUT -p tcp -j "$CHAIN4" 2>/dev/null || true

socks_edge.py

@@ -66,6 +66,24 @@ class RelayLink:
             await self.writer.wait_closed()
 
 
+@dataclass
+class UdpFlowState:
+    flow_id: int
+    client_addr: tuple[str, int]
+    target_host: str
+    target_port: int
+    created_at: float
+    last_activity: float
+    packets_sent: int = 0
+    packets_received: int = 0
+    duplicate_responses: int = 0
+    winner_name: str | None = None
+    candidate_names: tuple[str, ...] = ()
+
+    def touch(self, now: float) -> None:
+        self.last_activity = now
+
+
 @dataclass
 class TcpRaceSession:
     session_id: int
@@ -174,6 +192,11 @@ class UdpAssociateServer(asyncio.DatagramProtocol):
         self.client_addr = None
         self.packet_counter = itertools.count(1)
         self.pending: set[int] = set()
+        self.packet_flows: dict[int, int] = {}
+        self.client_flows: dict[tuple[tuple[str, int], str, int], UdpFlowState] = {}
+        self.flow_counter = itertools.count(1)
+        self.last_summary_at = 0.0
+        self.win_counts: Dict[str, int] = {}
 
     def connection_made(self, transport) -> None:
         self.transport = transport
@@ -183,21 +206,105 @@ class UdpAssociateServer(asyncio.DatagramProtocol):
             return
         if self.client_addr is None:
             self.client_addr = addr
+            print(f"[edge] udp client bound addr={addr[0]}:{addr[1]}")
         if addr != self.client_addr:
             return
         host, port, payload = self._parse_socks_udp(data)
+        loop = asyncio.get_running_loop()
+        now = loop.time()
+        flow_key = ((addr[0], addr[1]), host, port)
+        flow = self.client_flows.get(flow_key)
+        if flow is None:
+            flow = UdpFlowState(
+                flow_id=next(self.flow_counter),
+                client_addr=(addr[0], addr[1]),
+                target_host=host,
+                target_port=port,
+                created_at=now,
+                last_activity=now,
+            )
+            self.client_flows[flow_key] = flow
+        flow.touch(now)
+        flow.packets_sent += 1
         packet_id = next(self.packet_counter)
         self.pending.add(packet_id)
+        self.packet_flows[packet_id] = flow.flow_id
+        print(f"[edge] udp recv flow={flow.flow_id} packet_id={packet_id} target={host}:{port} size={len(payload)}")
         asyncio.create_task(self.edge.forward_udp(host, port, payload, packet_id, self))
+        self._log_udp_summary()
 
-    async def handle_from_relay(self, frame: Frame, _link: RelayLink) -> None:
+    async def handle_from_relay(self, frame: Frame, link: RelayLink) -> None:
         if frame.packet_id not in self.pending or self.transport is None or self.client_addr is None:
             return
         self.pending.discard(frame.packet_id)
+        flow_id = self.packet_flows.pop(frame.packet_id, 0)
         host = self.edge.udp_targets.get(frame.packet_id, ("0.0.0.0", 0))[0]
         port = self.edge.udp_targets.get(frame.packet_id, ("0.0.0.0", 0))[1]
         packet = self._build_socks_udp(host, port, frame.payload)
+        winner_log = ""
+        flow = self._find_flow(flow_id)
+        if flow is not None:
+            now = asyncio.get_running_loop().time()
+            flow.touch(now)
+            flow.packets_received += 1
+            if flow.winner_name is None:
+                flow.winner_name = link.node.name
+                self.win_counts[link.node.name] = self.win_counts.get(link.node.name, 0) + 1
+                relay_detail = ", ".join(f"{name}={count}" for name, count in sorted(self.win_counts.items())) or "none"
+                print(
+                    f"[edge] udp flow={flow.flow_id} winner={link.node.name} "
+                    f"target={flow.target_host}:{flow.target_port} mode=single candidates={len(flow.candidate_names) or len(self.edge.links)}"
+                )
+                print(f"[edge] udp win relay_breakdown={relay_detail}")
+            elif flow.winner_name != link.node.name:
+                flow.duplicate_responses += 1
+                winner_log = f" duplicate=1 winner={flow.winner_name} from={link.node.name}"
+        print(
+            f"[edge] udp send flow={flow_id or 'unknown'} packet_id={frame.packet_id} "
+            f"target={host}:{port} size={len(frame.payload)} relay={link.node.name}{winner_log}"
+        )
         self.transport.sendto(packet, self.client_addr)
+        self._log_udp_summary()
+
+    def set_flow_candidates(self, packet_id: int, candidate_names: tuple[str, ...]) -> None:
+        flow_id = self.packet_flows.get(packet_id)
+        flow = self._find_flow(flow_id)
+        if flow is not None and not flow.candidate_names:
+            flow.candidate_names = candidate_names
+
+    def note_unsent(self, packet_id: int) -> None:
+        flow_id = self.packet_flows.pop(packet_id, 0)
+        self.pending.discard(packet_id)
+        flow = self._find_flow(flow_id)
+        if flow is not None:
+            flow.touch(asyncio.get_running_loop().time())
+        print(f"[edge] udp drop flow={flow_id or 'unknown'} packet_id={packet_id} reason=no_available_links")
+        self._log_udp_summary(force=True)
+
+    def _find_flow(self, flow_id: int | None) -> UdpFlowState | None:
+        if not flow_id:
+            return None
+        for flow in self.client_flows.values():
+            if flow.flow_id == flow_id:
+                return flow
+        return None
+
+    def _log_udp_summary(self, force: bool = False) -> None:
+        now = asyncio.get_running_loop().time()
+        if not force and now - self.last_summary_at < 10:
+            return
+        self.last_summary_at = now
+        active_flows = len(self.client_flows)
+        winners = sum(1 for flow in self.client_flows.values() if flow.winner_name)
+        packets_sent = sum(flow.packets_sent for flow in self.client_flows.values())
+        packets_received = sum(flow.packets_received for flow in self.client_flows.values())
+        duplicates = sum(flow.duplicate_responses for flow in self.client_flows.values())
+        print(
+            f"[edge] udp summary bind={self.client_addr[0]}:{self.client_addr[1]} active_flows={active_flows} "
+            f"winner_flows={winners} packets_sent={packets_sent} packets_received={packets_received} dup={duplicates}"
+            if self.client_addr
+            else f"[edge] udp summary bind=unbound active_flows={active_flows} winner_flows={winners} packets_sent={packets_sent} packets_received={packets_received} dup={duplicates}"
+        )
 
     def _parse_socks_udp(self, packet: bytes) -> tuple[str, int, bytes]:
         atyp = packet[3]
@@ -261,7 +368,8 @@ class SocksEdge:
 
     async def _accept(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
         try:
-            host, port, udp_mode = await self._handshake(reader, writer)
+            peer = writer.get_extra_info("peername")
+            host, port, udp_mode = await self._handshake(reader, writer, peer)
             if udp_mode:
                 return
             links = self._selected_links()
@@ -290,11 +398,17 @@ class SocksEdge:
         self.udp_targets[packet_id] = (host, port)
         meta = encode_json({"host": host, "port": port})
         links = self._selected_links()
+        link_names = ",".join(link.node.name for link in links) or "none"
+        udp_server.set_flow_candidates(packet_id, tuple(link.node.name for link in links))
+        print(f"[edge] udp forward packet_id={packet_id} target={host}:{port} size={len(payload)} links={link_names}")
+        if not links:
+            udp_server.note_unsent(packet_id)
+            return
         for index, link in enumerate(links):
             body = meta + payload if index == 0 else payload
             await link.send(Frame(UDP_SEND, 1, index, 0, packet_id if index == 0 else 0, body))
 
-    async def _handshake(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> tuple[str, int, bool]:
+    async def _handshake(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter, peer) -> tuple[str, int, bool]:
         version, methods_len = (await read_exact(reader, 2))
         if version != SOCKS_VERSION:
             raise ValueError("unsupported socks version")
@@ -312,12 +426,15 @@ class SocksEdge:
         else:
             raise ValueError("unsupported atyp")
         port = struct.unpack("!H", await read_exact(reader, 2))[0]
+        peer_text = f"{peer[0]}:{peer[1]}" if isinstance(peer, tuple) and len(peer) >= 2 else str(peer)
         if command == 1:
+            print(f"[edge] socks handshake peer={peer_text} command=connect target={host}:{port}")
             writer.write(b"\x05\x00\x00\x01\x00\x00\x00\x00\x00\x00")
             await writer.drain()
             return host, port, False
         if command == 3 and self.udp_server and self.udp_server.transport:
             bind_host, bind_port = self.udp_server.transport.get_extra_info("sockname")[:2]
+            print(f"[edge] socks handshake peer={peer_text} command=udp_associate target={host}:{port} bind={bind_host}:{bind_port}")
             writer.write(b"\x05\x00\x00\x01" + socket.inet_aton(bind_host) + struct.pack("!H", bind_port))
             await writer.drain()
             return host, port, True

transparent_edge.py

@@ -424,38 +424,63 @@ class UdpFlow:
     target: TargetAddress
     send_response: Callable[[PeerAddress, bytes], Awaitable[None]]
     paths: list[BasePath]
+    redundancy: int = 0
+    always_broadcast: bool = True
+    copy_interval_ms: int = 0
     winner: BasePath | None = None
     closed: bool = False
     last_activity: float = 0.0
+    packets_sent: int = 0
+    packets_received: int = 0
+    duplicate_responses: int = 0
+    send_task: asyncio.Task | None = None
 
     async def start(self) -> None:
         await asyncio.gather(*(path.open(self.target) for path in self.paths), return_exceptions=True)
 
     async def send(self, payload: bytes) -> None:
         self.last_activity = asyncio.get_running_loop().time()
+        self.packets_sent += 1
         active = [path for path in self.paths if path.opened and not path.closed]
-        if self.winner is None:
-            await asyncio.gather(*(path.send(payload) for path in active), return_exceptions=True)
-        elif not self.winner.closed:
-            await self.winner.send(payload)
+        if not active:
+            return
+        copies = max(1, self.redundancy + 1)
+        targets = active if self.always_broadcast or self.winner is None or self.winner.closed else [self.winner]
+        for attempt in range(copies):
+            await asyncio.gather(*(path.send(payload) for path in targets), return_exceptions=True)
+            if attempt + 1 < copies and self.copy_interval_ms > 0:
+                await asyncio.sleep(self.copy_interval_ms / 1000)
 
     async def handle_path(self, path: BasePath, event: str, payload: bytes | None) -> None:
         self.last_activity = asyncio.get_running_loop().time()
         if event == "data" and payload is not None:
+            self.packets_received += 1
             if self.winner is None:
                 self.winner = path
-                print(f"[edge] udp flow={self.flow_id} winner={path.name} target={self.target.host}:{self.target.port}")
+                mode = "redundant" if self.redundancy > 0 else "single"
+                print(f"[edge] udp flow={self.flow_id} winner={path.name} target={self.target.host}:{self.target.port} mode={mode} candidates={len(self.paths)}")
+            elif path is not self.winner:
+                self.duplicate_responses += 1
             if path is self.winner:
                 await self.send_response(self.source, payload)
         if event == "close":
             path.closed = True
             if path is self.winner:
-                self.winner = None
+                remaining = [candidate for candidate in self.paths if candidate.opened and not candidate.closed]
+                self.winner = remaining[0] if remaining else None
 
     async def close(self) -> None:
         if self.closed:
             return
         self.closed = True
+        if self.send_task and self.send_task is not asyncio.current_task():
+            self.send_task.cancel()
+            with contextlib.suppress(Exception):
+                await self.send_task
+        print(
+            f"[edge] udp flow={self.flow_id} closed target={self.target.host}:{self.target.port} "
+            f"sent={self.packets_sent} received={self.packets_received} dup={self.duplicate_responses}"
+        )
         await asyncio.gather(*(path.close() for path in self.paths), return_exceptions=True)
 
 
@@ -466,6 +491,13 @@ class TransparentUdpListener:
         self.bind_host = bind_host
         self.port = port
         self.socket: socket.socket | None = None
+        self.udp_packets_received = 0
+        self.udp_recv_errors = 0
+        self.udp_parse_errors = 0
+        self.udp_missing_original = 0
+        self.udp_self_loop_skipped = 0
+        self.udp_flows_created = 0
+        self.last_summary_at = 0.0
 
     def start(self) -> None:
         sock = socket.socket(self.family, socket.SOCK_DGRAM)
@@ -481,29 +513,65 @@ class TransparentUdpListener:
         asyncio.get_running_loop().add_reader(sock.fileno(), self._on_readable)
         print(f"[edge] transparent udp listening on {sock.getsockname()}")
 
+    def _log_udp_summary(self, force: bool = False) -> None:
+        now = asyncio.get_running_loop().time()
+        if not force and now - self.last_summary_at < 10:
+            return
+        self.last_summary_at = now
+        print(
+            f"[edge] udp summary family={self.family} bind={self.bind_host}:{self.port} "
+            f"received={self.udp_packets_received} flows={self.udp_flows_created} "
+            f"self_loop={self.udp_self_loop_skipped} missing_original={self.udp_missing_original} "
+            f"parse_error={self.udp_parse_errors} recv_error={self.udp_recv_errors}"
+        )
+
     def _on_readable(self) -> None:
         assert self.socket is not None
         try:
             data, ancdata, _flags, src = self.socket.recvmsg(65535, 512)
         except BlockingIOError:
             return
-        except Exception:
+        except Exception as exc:
+            self.udp_recv_errors += 1
+            print(f"[edge] udp recvmsg error family={self.family} error={exc!r}")
+            self._log_udp_summary(force=True)
             return
+        self.udp_packets_received += 1
         original = None
         for level, ctype, cdata in ancdata:
             if self.family == socket.AF_INET and level == socket.SOL_IP and ctype == IP_RECVORIGDSTADDR:
-                original = parse_sockaddr(cdata)
+                try:
+                    original = parse_sockaddr(cdata)
+                except Exception as exc:
+                    self.udp_parse_errors += 1
+                    print(f"[edge] udp parse original dst error family={self.family} src={src} error={exc!r} raw_len={len(cdata)}")
+                    self._log_udp_summary(force=True)
+                    return
                 break
             if self.family == socket.AF_INET6 and level == socket.IPPROTO_IPV6 and ctype == IPV6_RECVORIGDSTADDR:
-                original = parse_sockaddr(cdata)
+                try:
+                    original = parse_sockaddr(cdata)
+                except Exception as exc:
+                    self.udp_parse_errors += 1
+                    print(f"[edge] udp parse original dst error family={self.family} src={src} error={exc!r} raw_len={len(cdata)}")
+                    self._log_udp_summary(force=True)
+                    return
                 break
         if original is None:
+            self.udp_missing_original += 1
+            self._log_udp_summary()
             return
         if self.family == socket.AF_INET:
             source = PeerAddress(host=src[0], port=src[1], family=socket.AF_INET)
         else:
             source = PeerAddress(host=src[0], port=src[1], family=socket.AF_INET6)
         if original.port == self.port and (original.host in ("127.0.0.1", "::1") or original.host == self.bind_host):
+            self.udp_self_loop_skipped += 1
+            print(
+                f"[edge] udp self_loop family={self.family} src={source.host}:{source.port} "
+                f"original={original.host}:{original.port} size={len(data)}"
+            )
+            self._log_udp_summary()
             return
         asyncio.create_task(self.edge.handle_udp_datagram(source, original, data, self))
 
@@ -617,6 +685,17 @@ class TransparentEdge:
             for index in range(count)
         ]
 
+    def _build_udp_direct_paths(self, target: TargetAddress, flow_id: int) -> list[BasePath]:
+        count = max(1, self.config.udp_direct_redundancy)
+        return [
+            DirectUdpPath(
+                name=f"direct-{index + 1}" if count > 1 else "direct",
+                on_frame=lambda path, event, data, fid=flow_id: self._handle_udp_path(fid, path, event, data),
+                target=target,
+            )
+            for index in range(count)
+        ]
+
     def _start_udp_listeners(self) -> None:
         binds = []
         if self.listen_host == "127.0.0.1":
@@ -678,13 +757,24 @@ class TransparentEdge:
         flow = self.udp_flows.get(key)
         if flow is None:
             flow_id = next(self.udp_flow_ids)
-            paths: list[BasePath] = [DirectUdpPath(name="direct", on_frame=lambda path, event, data, fid=flow_id: self._handle_udp_path(fid, path, event, data), target=target)]
+            paths: list[BasePath] = self._build_udp_direct_paths(target, flow_id)
             for connection in self.manager.available():
                 stream_id = next(self.stream_ids)
                 paths.append(RelayUdpPath(name=connection.node.name, on_frame=lambda path, event, data, fid=flow_id: self._handle_udp_path(fid, path, event, data), connection=connection, session_id=flow_id, stream_id=stream_id, target=target))
-            flow = UdpFlow(flow_id=flow_id, source=source, target=target, send_response=listener.send_response, paths=paths)
+            flow = UdpFlow(
+                flow_id=flow_id,
+                source=source,
+                target=target,
+                send_response=listener.send_response,
+                paths=paths,
+                redundancy=self.config.udp_redundancy,
+                always_broadcast=self.config.udp_always_broadcast,
+                copy_interval_ms=self.config.udp_copy_interval_ms,
+            )
             self.udp_flows[key] = flow
-            print(f"[edge] udp flow={flow_id} target={target.host}:{target.port} candidates={[path.name for path in paths]}")
+            listener.udp_flows_created += 1
+            listener._log_udp_summary(force=True)
+            print(f"[edge] udp flow={flow_id} source={source.host}:{source.port} target={target.host}:{target.port} redundancy={self.config.udp_redundancy} direct_redundancy={self.config.udp_direct_redundancy} always_broadcast={self.config.udp_always_broadcast} candidates={[path.name for path in paths]}")
             await flow.start()
         await flow.send(payload)