feat: enhance validation for Queue.Mark and add corresponding unit tests; refactor bypass mark application in dialers

src/config/methods.go

@@ -276,6 +276,11 @@ func (c *Config) Validate() error {
 		return fmt.Errorf("mark value 0x%x is too high for auto-derived discovery marks", c.Queue.Mark)
 	}
 
+	const perSetReachableBits uint32 = 0x17FFF
+	if c.Queue.Mark != 0 && uint32(c.Queue.Mark)&^perSetReachableBits == 0 {
+		return fmt.Errorf("mark value 0x%x conflicts with per-set mark bits {0-14, 16}; bypass rule would catch TPROXY-redirected traffic. Use a value with at least one bit in {15, 17-31} (default 0x8000 has bit 15)", c.Queue.Mark)
+	}
+
 	c.System.Checker.DiscoveryFlowMark = c.DiscoveryFlowMark()
 	c.System.Checker.DiscoveryInjectedMark = c.DiscoveryInjectedMark()
 

src/config/methods_test.go

@@ -102,6 +102,28 @@ func TestValidate(t *testing.T) {
 		}
 	})
 
+	t.Run("queue mark inside per-set space fails", func(t *testing.T) {
+		cases := []uint{0x4000, 0x100, 0x10000, 0x12345, 0x17DFF}
+		for _, m := range cases {
+			cfg := NewConfig()
+			cfg.Queue.Mark = m
+			if err := cfg.Validate(); err == nil {
+				t.Errorf("expected error for Queue.Mark=%#x (collides with per-set range)", m)
+			}
+		}
+	})
+
+	t.Run("queue mark outside per-set space passes", func(t *testing.T) {
+		cases := []uint{0x8000, 0x18000, 0x20000, 0x80000000}
+		for _, m := range cases {
+			cfg := NewConfig()
+			cfg.Queue.Mark = m
+			if err := cfg.Validate(); err != nil {
+				t.Errorf("Queue.Mark=%#x should pass, got: %v", m, err)
+			}
+		}
+	})
+
 	t.Run("queue num out of range", func(t *testing.T) {
 		cfg := NewConfig()
 		cfg.Queue.StartNum = -1

src/http/ui/src/components/dashboard/MetricsCards.tsx

@@ -11,8 +11,8 @@ interface MetricsCardsProps {
 export const MetricsCards = ({ metrics }: MetricsCardsProps) => {
   const { t } = useTranslation();
   const targetRate =
-    metrics.packets_processed > 0
-      ? ((metrics.targeted_connections / metrics.packets_processed) * 100).toFixed(1)
+    metrics.total_connections > 0
+      ? ((metrics.targeted_connections / metrics.total_connections) * 100).toFixed(1)
       : "0.0";
 
   const isIdle = metrics.rst_dropped === 0;

src/http/ws/connections.go

@@ -24,8 +24,8 @@ func HandleConnectionsWebSocket(w http.ResponseWriter, r *http.Request) {
 	ch, snapshot := hub.Subscribe()
 	defer hub.Unsubscribe(ch)
 
-	conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
 	for _, msg := range snapshot {
+		conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
 		if err := conn.WriteMessage(websocket.TextMessage, []byte(msg)); err != nil {
 			return
 		}
@@ -46,10 +46,7 @@ func HandleConnectionsWebSocket(w http.ResponseWriter, r *http.Request) {
 
 	for {
 		select {
-		case msg, ok := <-ch:
-			if !ok {
-				return
-			}
+		case msg := <-ch:
 			conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
 			if err := conn.WriteMessage(websocket.TextMessage, []byte(msg)); err != nil {
 				return

src/http/ws/discovery.go

@@ -36,10 +36,7 @@ func HandleDiscoveryWebSocket(w http.ResponseWriter, r *http.Request) {
 
 	for {
 		select {
-		case msg, ok := <-ch:
-			if !ok {
-				return
-			}
+		case msg := <-ch:
 			conn.SetWriteDeadline(time.Now().Add(10 * time.Second))
 			if err := conn.WriteMessage(websocket.TextMessage, []byte(msg)); err != nil {
 				return

src/log/connection.go

@@ -55,7 +55,6 @@ func (h *ConnectionHub) Unsubscribe(ch chan string) {
 	for i, l := range h.listeners {
 		if l == ch {
 			h.listeners = append(h.listeners[:i], h.listeners[i+1:]...)
-			close(ch)
 			return
 		}
 	}

src/log/discovery.go

@@ -38,7 +38,6 @@ func (h *DiscoveryLogHub) Unsubscribe(ch chan string) {
 	for i, l := range h.listeners {
 		if l == ch {
 			h.listeners = append(h.listeners[:i], h.listeners[i+1:]...)
-			close(ch)
 			return
 		}
 	}

src/socks5/client.go

@@ -25,6 +25,22 @@ type ClientConfig struct {
 	BypassMark uint32
 }
 
+func ApplyBypassMark(d *net.Dialer, mark uint32) {
+	if mark == 0 {
+		return
+	}
+	d.Control = func(network, address string, c syscall.RawConn) error {
+		var sockErr error
+		err := c.Control(func(fd uintptr) {
+			sockErr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_MARK, int(mark))
+		})
+		if err != nil {
+			return err
+		}
+		return sockErr
+	}
+}
+
 func DialUpstream(ctx context.Context, cfg ClientConfig, targetHost string, targetPort int) (net.Conn, error) {
 	if cfg.Host == "" || cfg.Port < 1 || cfg.Port > 65535 {
 		return nil, fmt.Errorf("invalid upstream config")
@@ -39,19 +55,7 @@ func DialUpstream(ctx context.Context, cfg ClientConfig, targetHost string, targ
 	}
 
 	d := net.Dialer{Timeout: timeout}
-	if cfg.BypassMark != 0 {
-		mark := cfg.BypassMark
-		d.Control = func(network, address string, c syscall.RawConn) error {
-			var sockErr error
-			err := c.Control(func(fd uintptr) {
-				sockErr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_MARK, int(mark))
-			})
-			if err != nil {
-				return err
-			}
-			return sockErr
-		}
-	}
+	ApplyBypassMark(&d, cfg.BypassMark)
 	addr := net.JoinHostPort(cfg.Host, strconv.Itoa(cfg.Port))
 	conn, err := d.DialContext(ctx, "tcp", addr)
 	if err != nil {

src/tproxy/listener.go

@@ -18,20 +18,7 @@ import (
 
 func markedDialer(timeout time.Duration, bypassMark uint32) net.Dialer {
 	d := net.Dialer{Timeout: timeout}
-	if bypassMark == 0 {
-		return d
-	}
-	mark := bypassMark
-	d.Control = func(network, address string, c syscall.RawConn) error {
-		var sockErr error
-		err := c.Control(func(fd uintptr) {
-			sockErr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_MARK, int(mark))
-		})
-		if err != nil {
-			return err
-		}
-		return sockErr
-	}
+	socks5.ApplyBypassMark(&d, bypassMark)
 	return d
 }