From 90c5d9420028e2857a20b703a26f5874f51773e1 Mon Sep 17 00:00:00 2001 From: Hiddify <114227601+hiddify-com@users.noreply.github.com> Date: Wed, 1 Apr 2026 18:04:11 +0000 Subject: [PATCH 01/13] feat(client): implement MultiResolver with Healthcheck and recovery --- client/client.go | 71 +++-- client/multi_resolver.go | 610 +++++++++++++++++++++++++++++++++++++++ dns/dns.go | 6 +- vaydns-client/main.go | 98 ++++--- 4 files changed, 715 insertions(+), 70 deletions(-) create mode 100644 client/multi_resolver.go diff --git a/client/client.go b/client/client.go index d1c3a1a..c3337d5 100644 --- a/client/client.go +++ b/client/client.go @@ -30,13 +30,13 @@ import ( "syscall" "time" - log "github.com/sirupsen/logrus" - utls "github.com/refraction-networking/utls" - "github.com/xtaci/kcp-go/v5" - "github.com/xtaci/smux" "github.com/net2share/vaydns/dns" "github.com/net2share/vaydns/noise" "github.com/net2share/vaydns/turbotunnel" + utls "github.com/refraction-networking/utls" + log "github.com/sirupsen/logrus" + "github.com/xtaci/kcp-go/v5" + "github.com/xtaci/smux" ) // Default timeouts for VayDNS mode. @@ -47,10 +47,10 @@ const ( DefaultReconnectDelay = 1 * time.Second DefaultReconnectMaxDelay = 30 * time.Second DefaultSessionCheckInterval = 20 * time.Second - DefaultUDPResponseTimeout = 500 * time.Millisecond - DefaultUDPWorkers = 100 - DefaultMaxStreams = 256 - DefaultHandshakeTimeout = 15 * time.Second + DefaultUDPResponseTimeout = 500 * time.Millisecond + DefaultUDPWorkers = 100 + DefaultMaxStreams = 256 + DefaultHandshakeTimeout = 15 * time.Second ) // Default timeouts for dnstt compatibility mode. @@ -190,18 +190,18 @@ func (ts *TunnelServer) effectiveMaxQnameLen() int { // either call the step-by-step Initiate* methods (for embedding in frameworks // like xray-core) or call ListenAndServe for a fully managed session. type Tunnel struct { - Resolver Resolver + Resolvers []Resolver TunnelServer TunnelServer // Session configuration. Zero values use defaults. - IdleTimeout time.Duration // default: 10s (2m with DnsttCompat) - KeepAlive time.Duration // default: 2s (10s with DnsttCompat) - OpenStreamTimeout time.Duration // default: 10s - MaxStreams int // default: 256 (0 = unlimited) - ReconnectMinDelay time.Duration // default: 1s - ReconnectMaxDelay time.Duration // default: 30s - SessionCheckInterval time.Duration // default: 20s - HandshakeTimeout time.Duration // default: 30s + IdleTimeout time.Duration // default: 10s (2m with DnsttCompat) + KeepAlive time.Duration // default: 2s (10s with DnsttCompat) + OpenStreamTimeout time.Duration // default: 10s + MaxStreams int // default: 256 (0 = unlimited) + ReconnectMinDelay time.Duration // default: 1s + ReconnectMaxDelay time.Duration // default: 30s + SessionCheckInterval time.Duration // default: 20s + HandshakeTimeout time.Duration // default: 30s PacketQueueSize int // default: QueueSize (512) KCPWindowSize int // default: PacketQueueSize/2 QueueOverflowMode turbotunnel.QueueOverflowMode // default: drop @@ -219,9 +219,9 @@ type Tunnel struct { // NewTunnel creates a Tunnel with the given resolver and server configuration. // Zero-value fields use sensible defaults. -func NewTunnel(resolver Resolver, tunnelServer TunnelServer) (*Tunnel, error) { +func NewTunnel(resolvers []Resolver, tunnelServer TunnelServer) (*Tunnel, error) { t := &Tunnel{ - Resolver: resolver, + Resolvers: resolvers, TunnelServer: tunnelServer, } t.wireConfig = tunnelServer.wireConfig() @@ -293,7 +293,16 @@ func (t *Tunnel) effectiveKCPWindowSize() int { // InitiateResolverConnection creates the underlying transport connection // based on the Resolver configuration. func (t *Tunnel) InitiateResolverConnection() error { - r := t.Resolver + if len(t.Resolvers) > 1 { + conn, err := NewMultiResolver(t.Resolvers, SelectionRoundRobin, t.effectivePacketQueueSize(), t.effectiveQueueOverflowMode()) + if err != nil { + return err + } + t.resolverConn = conn + t.remoteAddr = turbotunnel.DummyAddr{} + return nil + } + r := t.Resolvers[0] switch r.ResolverType { case ResolverTypeUDP: addr, err := net.ResolveUDPAddr("udp", r.ResolverAddr) @@ -589,6 +598,24 @@ func (t *Tunnel) closeTransportLayers() { t.forgedStats = nil } +// MultiResolverStats returns per-resolver health and count snapshots when the +// active resolver transport is MultiResolver; otherwise it returns nil. +func (t *Tunnel) MultiResolverStats() []ResolverStat { + if mr, ok := t.resolverConn.(*MultiResolver); ok { + return mr.ResolverStats() + } + return nil +} + +// MultiResolverValidInvalidCounts returns valid/invalid counters per resolver +// address when the active resolver transport is MultiResolver. +func (t *Tunnel) MultiResolverValidInvalidCounts() map[string][2]int64 { + if mr, ok := t.resolverConn.(*MultiResolver); ok { + return mr.ValidInvalidCounts() + } + return nil +} + // resetTransportLayers tears down existing transport layers and creates fresh // ones. Used during reconnect to ensure a clean transport stack. func (t *Tunnel) resetTransportLayers() error { @@ -882,10 +909,10 @@ func NewOutbound(resolvers []Resolver, tunnelServers []TunnelServer) *Outbound { // Start begins accepting connections on bind and forwarding them through the // first resolver/server pair. func (o *Outbound) Start(bind string) error { - resolver := o.Resolvers[0] + tunnelServer := o.TunnelServers[0] - tunnel, err := NewTunnel(resolver, tunnelServer) + tunnel, err := NewTunnel(o.Resolvers, tunnelServer) if err != nil { return fmt.Errorf("failed to create tunnel: %w", err) } diff --git a/client/multi_resolver.go b/client/multi_resolver.go new file mode 100644 index 0000000..fdba9cf --- /dev/null +++ b/client/multi_resolver.go @@ -0,0 +1,610 @@ +package client + +import ( + "context" + "crypto/tls" + "fmt" + "net" + "net/http" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/net2share/vaydns/dns" + "github.com/net2share/vaydns/turbotunnel" + log "github.com/sirupsen/logrus" +) + +// SelectionMode controls which resolver MultiResolver picks for outgoing packets. +type SelectionMode string + +const ( + // SelectionRoundRobin rotates through preferred resolvers in order. + SelectionRoundRobin SelectionMode = "roundrobin" + // SelectionBest picks the resolver with the best observed score. + SelectionBest SelectionMode = "best" + // SelectionSmart currently aliases best-score selection. + SelectionSmart SelectionMode = "smart" +) + +// ResolverState is the current health state of one resolver. +type ResolverState string + +const ( + ResolverStateUnknown ResolverState = "unknown" + ResolverStateHealthy ResolverState = "healthy" + ResolverStateRateLimited ResolverState = "rate_limited" + ResolverStateDown ResolverState = "down" +) + +const ( + pendingResponseTimeout = 5 * time.Second + healthTickInterval = 1 * time.Second + downTimeoutThreshold = int64(8) + rateLimitThreshold = int64(5) + probeInterval = 1 * time.Second +) + +// ResolverStat is a snapshot of one resolver's counters and health state. +type ResolverStat struct { + Address string + State ResolverState + ValidCount int64 + InvalidCount int64 + TimeoutCount int64 + LastWrite time.Time + LastValid time.Time +} + +type resolverEntry struct { + name string + addr net.Addr + conn net.PacketConn + + validCount atomic.Int64 + invalidCount atomic.Int64 + timeoutCount atomic.Int64 + + mu sync.Mutex + pending map[uint16]time.Time + lastWrite time.Time + lastValid time.Time + lastProbe time.Time + state ResolverState +} + +func (e *resolverEntry) writePacket(b []byte) (int, error) { + now := time.Now() + e.trackOutgoingID(b, now) + n, err := e.conn.WriteTo(b, e.addr) + if err != nil { + e.invalidCount.Add(1) + } + e.mu.Lock() + e.lastWrite = now + e.mu.Unlock() + return n, err +} + +func (e *resolverEntry) trackOutgoingID(b []byte, now time.Time) { + msg, err := dns.MessageFromWireFormat(b) + if err != nil { + return + } + e.mu.Lock() + e.pending[msg.ID] = now + e.mu.Unlock() +} + +func (e *resolverEntry) readPacket() multiReadResult { + var result multiReadResult + result.entry = e + result.n, result.addr, result.err = e.conn.ReadFrom(result.buf[:]) + if result.err == nil { + e.evaluateIncoming(result.buf[:result.n]) + } + return result +} + +func (e *resolverEntry) evaluateIncoming(packet []byte) { + resp, err := dns.MessageFromWireFormat(packet) + if err != nil { + e.invalidCount.Add(1) + e.recomputeState(time.Now()) + return + } + + e.mu.Lock() + delete(e.pending, resp.ID) + e.mu.Unlock() + + if isValidDNSResponse(resp) { + e.validCount.Add(1) + e.timeoutCount.Store(0) + e.mu.Lock() + e.lastValid = time.Now() + e.state = ResolverStateHealthy + e.mu.Unlock() + return + } + + e.invalidCount.Add(1) + if isRateLimitedResponse(resp) { + e.mu.Lock() + e.state = ResolverStateRateLimited + e.mu.Unlock() + } + e.recomputeState(time.Now()) +} + +func (e *resolverEntry) expirePending(now time.Time) { + expired := int64(0) + e.mu.Lock() + for id, t := range e.pending { + if now.Sub(t) >= pendingResponseTimeout { + delete(e.pending, id) + expired++ + } + } + e.mu.Unlock() + if expired > 0 { + e.timeoutCount.Add(expired) + e.invalidCount.Add(expired) + } + e.recomputeState(now) +} + +func (e *resolverEntry) recomputeState(now time.Time) { + e.mu.Lock() + defer e.mu.Unlock() + + timeouts := e.timeoutCount.Load() + invalid := e.invalidCount.Load() + valid := e.validCount.Load() + + switch { + case timeouts >= downTimeoutThreshold: + e.state = ResolverStateDown + case invalid >= rateLimitThreshold && valid == 0: + e.state = ResolverStateRateLimited + case valid > 0 && now.Sub(e.lastValid) <= 30*time.Second: + e.state = ResolverStateHealthy + case valid == 0: + e.state = ResolverStateUnknown + default: + e.state = ResolverStateUnknown + } + + // Slow decay to avoid sticky penalties. + if invalid > 0 { + e.invalidCount.Store(invalid - 1) + } + if timeouts > 0 { + e.timeoutCount.Store(timeouts - 1) + } +} + +func (e *resolverEntry) stateSnapshot() ResolverState { + e.mu.Lock() + defer e.mu.Unlock() + return e.state +} + +func (e *resolverEntry) markProbe(now time.Time) { + e.mu.Lock() + e.lastProbe = now + e.mu.Unlock() +} + +func (e *resolverEntry) canProbe(now time.Time) bool { + e.mu.Lock() + defer e.mu.Unlock() + return now.Sub(e.lastProbe) >= probeInterval +} + +func (e *resolverEntry) snapshot() ResolverStat { + e.mu.Lock() + defer e.mu.Unlock() + return ResolverStat{ + Address: e.name, + State: e.state, + ValidCount: e.validCount.Load(), + InvalidCount: e.invalidCount.Load(), + TimeoutCount: e.timeoutCount.Load(), + LastWrite: e.lastWrite, + LastValid: e.lastValid, + } +} + +type multiReadResult struct { + buf [4096]byte + n int + addr net.Addr + err error + entry *resolverEntry +} + +// MultiResolver is a net.PacketConn that multiplexes across multiple DNS +// resolver transport connections. It tracks per-resolver health from valid and +// invalid responses, avoids down resolvers for primary traffic, and probes +// unhealthy resolvers by duplicating selected packets. +type MultiResolver struct { + entries []*resolverEntry + mode SelectionMode + mu sync.Mutex + rrIndex int + probeRR int + recvChan chan multiReadResult + closed chan struct{} + closeOnce sync.Once +} + +// NewMultiResolver creates a MultiResolver from a slice of Resolver configs. +func NewMultiResolver(resolvers []Resolver, mode SelectionMode, queueSize int, overflowMode turbotunnel.QueueOverflowMode) (*MultiResolver, error) { + if len(resolvers) == 0 { + return nil, fmt.Errorf("at least one resolver is required") + } + + entries := make([]*resolverEntry, 0, len(resolvers)) + for _, r := range resolvers { + conn, addr, err := getResolverConnection(r, queueSize, overflowMode) + if err != nil { + for _, e := range entries { + e.conn.Close() + } + return nil, fmt.Errorf("resolver %s %s: %w", r.ResolverType, r.ResolverAddr, err) + } + entries = append(entries, &resolverEntry{ + name: r.ResolverAddr, + addr: addr, + conn: conn, + pending: make(map[uint16]time.Time), + state: ResolverStateUnknown, + }) + } + + mr := &MultiResolver{ + entries: entries, + mode: mode, + recvChan: make(chan multiReadResult, len(entries)*4), + closed: make(chan struct{}), + } + for _, e := range entries { + entry := e + go func() { + for { + res := entry.readPacket() + select { + case mr.recvChan <- res: + case <-mr.closed: + return + } + if res.err != nil { + return + } + } + }() + } + go mr.healthWorker() + return mr, nil +} + +func (mr *MultiResolver) healthWorker() { + ticker := time.NewTicker(healthTickInterval) + defer ticker.Stop() + for { + select { + case <-ticker.C: + now := time.Now() + stats := make([]ResolverStat, 0, len(mr.entries)) + for _, e := range mr.entries { + e.expirePending(now) + stats = append(stats, e.snapshot()) + } + log.Trace("\n" + renderResolverStatsTable(stats, now)) + case <-mr.closed: + return + } + } +} + +func renderResolverStatsTable(stats []ResolverStat, now time.Time) string { + var b strings.Builder + b.WriteString("+-------------------------+--------------+--------+---------+---------+-------------+-------------+\n") + b.WriteString("| resolver | state | valid | invalid | timeout | last_write | last_valid |\n") + b.WriteString("+-------------------------+--------------+--------+---------+---------+-------------+-------------+\n") + for _, s := range stats { + lastWriteAgo := "-" + if !s.LastWrite.IsZero() { + lastWriteAgo = now.Sub(s.LastWrite).Truncate(time.Second).String() + } + lastValidAgo := "-" + if !s.LastValid.IsZero() { + lastValidAgo = now.Sub(s.LastValid).Truncate(time.Second).String() + } + b.WriteString(fmt.Sprintf("| %-23.23s | %-12s | %6d | %7d | %7d | %11s | %11s |\n", + s.Address, + s.State, + s.ValidCount, + s.InvalidCount, + s.TimeoutCount, + lastWriteAgo, + lastValidAgo, + )) + } + b.WriteString("+-------------------------+--------------+--------+---------+---------+-------------+-------------+") + return b.String() +} + +func isValidDNSResponse(resp dns.Message) bool { + if resp.Flags&0x8000 == 0 { + return false + } + return (resp.Flags & 0x000f) == dns.RcodeNoError +} + +func isRateLimitedResponse(resp dns.Message) bool { + rcode := resp.Flags & 0x000f + return rcode == dns.RcodeRefused || rcode == dns.RcodeServerFailure +} + +// ReadFrom receives a packet from whichever resolver responds first. +func (mr *MultiResolver) ReadFrom(b []byte) (n int, addr net.Addr, err error) { + select { + case <-mr.closed: + return 0, nil, net.ErrClosed + case res := <-mr.recvChan: + if res.err != nil { + return 0, res.addr, res.err + } + n = copy(b, res.buf[:res.n]) + return n, turbotunnel.DummyAddr{}, nil + } +} + +// WriteTo sends b to the selected primary resolver and may duplicate b to one +// unhealthy resolver as a probe to detect recovery. +func (mr *MultiResolver) WriteTo(b []byte, _ net.Addr) (n int, err error) { + select { + case <-mr.closed: + return 0, net.ErrClosed + default: + } + + primary := mr.selectPrimary() + n, err = primary.writePacket(b) + if err != nil { + return n, err + } + + if probe := mr.selectProbeTarget(primary); probe != nil { + probe.markProbe(time.Now()) + _, _ = probe.writePacket(b) + } + return n, nil +} + +func (mr *MultiResolver) selectPrimary() *resolverEntry { + if mr.mode == SelectionRoundRobin { + if e := mr.selectRoundRobinHealthy(); e != nil { + return e + } + return mr.entries[0] + } + if e := mr.selectBestScore(); e != nil { + return e + } + return mr.entries[0] +} + +func (mr *MultiResolver) selectRoundRobinHealthy() *resolverEntry { + mr.mu.Lock() + defer mr.mu.Unlock() + + if len(mr.entries) == 0 { + return nil + } + + start := mr.rrIndex + for i := 0; i < len(mr.entries); i++ { + idx := (start + i) % len(mr.entries) + state := mr.entries[idx].stateSnapshot() + if state == ResolverStateHealthy || state == ResolverStateUnknown { + mr.rrIndex = (idx + 1) % len(mr.entries) + return mr.entries[idx] + } + } + idx := start % len(mr.entries) + mr.rrIndex = (idx + 1) % len(mr.entries) + return mr.entries[idx] +} + +func (mr *MultiResolver) selectBestScore() *resolverEntry { + if len(mr.entries) == 0 { + return nil + } + best := mr.entries[0] + bestScore := resolverScore(best) + for _, e := range mr.entries[1:] { + s := resolverScore(e) + if s > bestScore { + best = e + bestScore = s + } + } + return best +} + +func resolverScore(e *resolverEntry) int64 { + statePenalty := int64(0) + switch e.stateSnapshot() { + case ResolverStateHealthy: + statePenalty = 0 + case ResolverStateUnknown: + statePenalty = 5 + case ResolverStateRateLimited: + statePenalty = 15 + case ResolverStateDown: + statePenalty = 30 + } + return e.validCount.Load()*4 - e.invalidCount.Load()*2 - e.timeoutCount.Load()*3 - statePenalty +} + +func (mr *MultiResolver) selectProbeTarget(primary *resolverEntry) *resolverEntry { + mr.mu.Lock() + defer mr.mu.Unlock() + + now := time.Now() + for i := 0; i < len(mr.entries); i++ { + idx := (mr.probeRR + i) % len(mr.entries) + e := mr.entries[idx] + if e == primary { + continue + } + state := e.stateSnapshot() + if state == ResolverStateHealthy { + continue + } + if e.canProbe(now) { + mr.probeRR = (idx + 1) % len(mr.entries) + return e + } + } + return nil +} + +// ResolverStats returns current resolver health counters. +func (mr *MultiResolver) ResolverStats() []ResolverStat { + stats := make([]ResolverStat, 0, len(mr.entries)) + for _, e := range mr.entries { + stats = append(stats, e.snapshot()) + } + return stats +} + +// ValidInvalidCounts returns valid/invalid counts by resolver address. +func (mr *MultiResolver) ValidInvalidCounts() map[string][2]int64 { + out := make(map[string][2]int64, len(mr.entries)) + for _, e := range mr.entries { + out[e.name] = [2]int64{e.validCount.Load(), e.invalidCount.Load()} + } + return out +} + +// Close closes all underlying connections and stops the reader goroutines. +func (mr *MultiResolver) Close() error { + mr.closeOnce.Do(func() { + close(mr.closed) + for _, e := range mr.entries { + e.conn.Close() + } + }) + return nil +} + +// LocalAddr returns the local address of the first underlying connection. +func (mr *MultiResolver) LocalAddr() net.Addr { + return mr.entries[0].conn.LocalAddr() +} + +// SetDeadline sets a deadline on all underlying connections. +func (mr *MultiResolver) SetDeadline(t time.Time) error { + var last error + for _, e := range mr.entries { + if err := e.conn.SetDeadline(t); err != nil { + last = err + } + } + return last +} + +// SetReadDeadline sets a read deadline on all underlying connections. +func (mr *MultiResolver) SetReadDeadline(t time.Time) error { + var last error + for _, e := range mr.entries { + if err := e.conn.SetReadDeadline(t); err != nil { + last = err + } + } + return last +} + +// SetWriteDeadline sets a write deadline on all underlying connections. +func (mr *MultiResolver) SetWriteDeadline(t time.Time) error { + var last error + for _, e := range mr.entries { + if err := e.conn.SetWriteDeadline(t); err != nil { + last = err + } + } + return last +} + +// getResolverConnection creates the underlying transport net.PacketConn for r. +func getResolverConnection(r Resolver, queueSize int, overflowMode turbotunnel.QueueOverflowMode) (net.PacketConn, net.Addr, error) { + switch r.ResolverType { + case ResolverTypeUDP: + addr, err := net.ResolveUDPAddr("udp", r.ResolverAddr) + if err != nil { + return nil, nil, err + } + if r.UDPSharedSocket { + lc := net.ListenConfig{Control: r.DialerControl} + conn, err := lc.ListenPacket(context.Background(), "udp", ":0") + if err != nil { + return nil, nil, err + } + return conn, addr, nil + } + workers := r.UDPWorkers + if workers <= 0 { + workers = DefaultUDPWorkers + } + timeout := r.UDPTimeout + if timeout <= 0 { + timeout = DefaultUDPResponseTimeout + } + conn, _, err := NewUDPPacketConn(addr, r.DialerControl, workers, timeout, !r.UDPAcceptErrors, queueSize, overflowMode) + if err != nil { + return nil, nil, err + } + return conn, addr, nil + + case ResolverTypeDOH: + var rt http.RoundTripper + if r.RoundTripper != nil { + rt = r.RoundTripper + } else if r.UTLSClientHelloID != nil { + rt = NewUTLSRoundTripper(nil, r.UTLSClientHelloID) + } else { + rt = http.DefaultTransport + } + conn, err := NewHTTPPacketConn(rt, r.ResolverAddr, 8, queueSize, overflowMode) + if err != nil { + return nil, nil, err + } + return conn, turbotunnel.DummyAddr{}, nil + + case ResolverTypeDOT: + var dialTLSContext func(ctx context.Context, network, addr string) (net.Conn, error) + if r.UTLSClientHelloID != nil { + id := r.UTLSClientHelloID + dialTLSContext = func(ctx context.Context, network, addr string) (net.Conn, error) { + return UTLSDialContext(ctx, network, addr, nil, id) + } + } else { + dialTLSContext = func(ctx context.Context, network, addr string) (net.Conn, error) { + return tls.DialWithDialer(&net.Dialer{}, network, addr, nil) + } + } + conn, err := NewTLSPacketConn(r.ResolverAddr, dialTLSContext, queueSize, overflowMode) + if err != nil { + return nil, nil, err + } + return conn, turbotunnel.DummyAddr{}, nil + + default: + return nil, nil, fmt.Errorf("unsupported resolver type: %s", r.ResolverType) + } +} diff --git a/dns/dns.go b/dns/dns.go index 8f35a6a..a6d0a62 100644 --- a/dns/dns.go +++ b/dns/dns.go @@ -47,8 +47,9 @@ var ( const ( // https://tools.ietf.org/html/rfc1035#section-3.2.2 - RRTypeA = 1 - RRTypeNS = 2 + RRTypeA = 1 + RRTypeNS = 2 + RRTypeCNAME = 5 RRTypeMX = 15 RRTypeTXT = 16 @@ -66,6 +67,7 @@ const ( RcodeServerFailure = 2 // a.k.a. SERVFAIL RcodeNameError = 3 // a.k.a. NXDOMAIN RcodeNotImplemented = 4 // a.k.a. NOTIMPL + RcodeRefused = 5 // a.k.a. REFUSED // https://tools.ietf.org/html/rfc6891#section-9 ExtendedRcodeBadVers = 16 // a.k.a. BADVERS ) diff --git a/vaydns-client/main.go b/vaydns-client/main.go index f6e9ac5..65bb893 100644 --- a/vaydns-client/main.go +++ b/vaydns-client/main.go @@ -20,6 +20,16 @@ import ( log "github.com/sirupsen/logrus" ) +type StringSliceFlag []string + +func (s *StringSliceFlag) String() string { + return fmt.Sprint(*s) +} + +func (s *StringSliceFlag) Set(value string) error { + *s = append(*s, value) + return nil +} func readKeyFromFile(filename string) ([]byte, error) { f, err := os.Open(filename) if err != nil { @@ -30,13 +40,13 @@ func readKeyFromFile(filename string) ([]byte, error) { } func main() { - var dohURL string - var dotAddr string + var dohURLs StringSliceFlag + var dotAddrs StringSliceFlag var domainArg string var listenAddr string var pubkeyFilename string var pubkeyString string - var udpAddr string + var udpAddrs StringSliceFlag var utlsDistribution string var maxQnameLen int var maxNumLabels int @@ -91,11 +101,11 @@ Known TLS fingerprints for -utls are: fmt.Fprintln(flag.CommandLine.Output(), line.String()) } } - flag.StringVar(&dohURL, "doh", "", "URL of DoH resolver") - flag.StringVar(&dotAddr, "dot", "", "address of DoT resolver") + flag.Var(&dohURLs, "doh", "URL of DoH resolver") + flag.Var(&dotAddrs, "dot", "address of DoT resolver") flag.StringVar(&pubkeyString, "pubkey", "", fmt.Sprintf("server public key (%d hex digits)", noise.KeyLen*2)) flag.StringVar(&pubkeyFilename, "pubkey-file", "", "read server public key from file") - flag.StringVar(&udpAddr, "udp", "", "address of UDP DNS resolver") + flag.Var(&udpAddrs, "udp", "address of UDP DNS resolver") flag.StringVar(&utlsDistribution, "utls", "4*random,3*Firefox_120,1*Firefox_105,3*Chrome_120,1*Chrome_102,1*iOS_14,1*iOS_13", "choose TLS fingerprint from weighted distribution") @@ -185,32 +195,42 @@ Known TLS fingerprints for -utls are: if utlsClientHelloID != nil { log.Infof("uTLS fingerprint %s %s", utlsClientHelloID.Client, utlsClientHelloID.Version) } - - // Select resolver transport. - var resolverType client.ResolverType - var resolverAddr string - transportCount := 0 - if dohURL != "" { - resolverType = client.ResolverTypeDOH - resolverAddr = dohURL - transportCount++ - } - if dotAddr != "" { - resolverType = client.ResolverTypeDOT - resolverAddr = dotAddr - transportCount++ - } - if udpAddr != "" { - resolverType = client.ResolverTypeUDP - resolverAddr = udpAddr - transportCount++ - } - if transportCount == 0 { - fmt.Fprintf(os.Stderr, "one of -doh, -dot, or -udp is required\n") + udpTimeout, err := time.ParseDuration(udpTimeoutStr) + if err != nil { + fmt.Fprintf(os.Stderr, "invalid -udp-timeout: %v\n", err) os.Exit(1) } - if transportCount > 1 { - fmt.Fprintf(os.Stderr, "only one of -doh, -dot, and -udp may be given\n") + // Select resolver transport. + resolvers := make([]client.Resolver, 0) + + for _, dohURL := range dohURLs { + resolver := client.Resolver{ + ResolverType: client.ResolverTypeDOH, + ResolverAddr: dohURL, + } + resolver.UTLSClientHelloID = utlsClientHelloID + resolvers = append(resolvers, resolver) + } + for _, dotAddr := range dotAddrs { + resolvers = append(resolvers, client.Resolver{ + ResolverType: client.ResolverTypeDOT, + ResolverAddr: dotAddr, + }) + } + for _, udpAddr := range udpAddrs { + resolver := client.Resolver{ + ResolverType: client.ResolverTypeUDP, + ResolverAddr: udpAddr, + } + resolver.UDPWorkers = udpWorkers + resolver.UDPSharedSocket = udpSharedSocket + resolver.UDPTimeout = udpTimeout + resolver.UDPAcceptErrors = udpAcceptErrors + resolvers = append(resolvers, resolver) + } + + if len(resolvers) == 0 { + fmt.Fprintf(os.Stderr, "one of -doh, -dot, or -udp is required\n") os.Exit(1) } @@ -245,11 +265,6 @@ Known TLS fingerprints for -utls are: fmt.Fprintf(os.Stderr, "invalid -open-stream-timeout: %v\n", err) os.Exit(1) } - udpTimeout, err := time.ParseDuration(udpTimeoutStr) - if err != nil { - fmt.Fprintf(os.Stderr, "invalid -udp-timeout: %v\n", err) - os.Exit(1) - } // Validate. if keepAlive >= idleTimeout { @@ -323,16 +338,7 @@ Known TLS fingerprints for -utls are: } // Build resolver. - resolver, err := client.NewResolver(resolverType, resolverAddr) - if err != nil { - fmt.Fprintf(os.Stderr, "resolver: %v\n", err) - os.Exit(1) - } - resolver.UTLSClientHelloID = utlsClientHelloID - resolver.UDPWorkers = udpWorkers - resolver.UDPSharedSocket = udpSharedSocket - resolver.UDPTimeout = udpTimeout - resolver.UDPAcceptErrors = udpAcceptErrors + if udpAcceptErrors { if udpSharedSocket { log.Warnf("-udp-accept-errors has no effect when -udp-shared-socket is set") @@ -355,7 +361,7 @@ Known TLS fingerprints for -utls are: ts.RecordType = recordTypeStr // Build tunnel. - tunnel, err := client.NewTunnel(resolver, ts) + tunnel, err := client.NewTunnel(resolvers, ts) if err != nil { fmt.Fprintf(os.Stderr, "%v\n", err) os.Exit(1) From 1acb5b3dfb0b2141e77a93622811dd4431ce3d03 Mon Sep 17 00:00:00 2001 From: Hiddify <114227601+hiddify-com@users.noreply.github.com> Date: Wed, 1 Apr 2026 22:20:03 +0000 Subject: [PATCH 02/13] refactor --- client/client.go | 80 +++------------------------------------- client/multi_resolver.go | 4 +- 2 files changed, 8 insertions(+), 76 deletions(-) diff --git a/client/client.go b/client/client.go index 5f96fae..0e1ffad 100644 --- a/client/client.go +++ b/client/client.go @@ -19,8 +19,6 @@ package client import ( - "context" - "crypto/tls" "errors" "fmt" "io" @@ -302,79 +300,13 @@ func (t *Tunnel) InitiateResolverConnection() error { t.remoteAddr = turbotunnel.DummyAddr{} return nil } - r := t.Resolvers[0] - switch r.ResolverType { - case ResolverTypeUDP: - addr, err := net.ResolveUDPAddr("udp", r.ResolverAddr) - if err != nil { - return err - } - t.remoteAddr = addr - if r.UDPSharedSocket { - lc := net.ListenConfig{Control: r.DialerControl} - conn, err := lc.ListenPacket(context.Background(), "udp", ":0") - if err != nil { - return err - } - t.resolverConn = conn - } else { - workers := r.UDPWorkers - if workers <= 0 { - workers = DefaultUDPWorkers - } - timeout := r.UDPTimeout - if timeout <= 0 { - timeout = DefaultUDPResponseTimeout - } - conn, forgedStats, err := NewUDPPacketConn(addr, r.DialerControl, workers, timeout, !r.UDPAcceptErrors, t.effectivePacketQueueSize(), t.effectiveQueueOverflowMode()) - if err != nil { - return err - } - t.forgedStats = forgedStats - t.resolverConn = conn - } - return nil - - case ResolverTypeDOH: - t.remoteAddr = turbotunnel.DummyAddr{} - var rt http.RoundTripper - if r.RoundTripper != nil { - rt = r.RoundTripper - } else if r.UTLSClientHelloID != nil { - rt = NewUTLSRoundTripper(nil, r.UTLSClientHelloID) - } else { - rt = http.DefaultTransport - } - conn, err := NewHTTPPacketConn(rt, r.ResolverAddr, 8, t.effectivePacketQueueSize(), t.effectiveQueueOverflowMode()) - if err != nil { - return err - } - t.resolverConn = conn - return nil - - case ResolverTypeDOT: - t.remoteAddr = turbotunnel.DummyAddr{} - var dialTLSContext func(ctx context.Context, network, addr string) (net.Conn, error) - if r.UTLSClientHelloID != nil { - id := r.UTLSClientHelloID - dialTLSContext = func(ctx context.Context, network, addr string) (net.Conn, error) { - return UTLSDialContext(ctx, network, addr, nil, id) - } - } else { - dialTLSContext = func(ctx context.Context, network, addr string) (net.Conn, error) { - return tls.DialWithDialer(&net.Dialer{}, network, addr, nil) - } - } - conn, err := NewTLSPacketConn(r.ResolverAddr, dialTLSContext, t.effectivePacketQueueSize(), t.effectiveQueueOverflowMode()) - if err != nil { - return err - } - t.resolverConn = conn - return nil - - default: - return fmt.Errorf("unsupported resolver type: %s", r.ResolverType) + conn, addr, err := GetResolverConnection(t.Resolvers[0], t.effectivePacketQueueSize(), t.effectiveQueueOverflowMode()) + if err != nil { + return err } + t.resolverConn = conn + t.remoteAddr = addr + return nil } // InitiateDNSPacketConn wraps the resolver connection with DNS encoding. diff --git a/client/multi_resolver.go b/client/multi_resolver.go index fdba9cf..c46f791 100644 --- a/client/multi_resolver.go +++ b/client/multi_resolver.go @@ -248,7 +248,7 @@ func NewMultiResolver(resolvers []Resolver, mode SelectionMode, queueSize int, o entries := make([]*resolverEntry, 0, len(resolvers)) for _, r := range resolvers { - conn, addr, err := getResolverConnection(r, queueSize, overflowMode) + conn, addr, err := GetResolverConnection(r, queueSize, overflowMode) if err != nil { for _, e := range entries { e.conn.Close() @@ -542,7 +542,7 @@ func (mr *MultiResolver) SetWriteDeadline(t time.Time) error { } // getResolverConnection creates the underlying transport net.PacketConn for r. -func getResolverConnection(r Resolver, queueSize int, overflowMode turbotunnel.QueueOverflowMode) (net.PacketConn, net.Addr, error) { +func GetResolverConnection(r Resolver, queueSize int, overflowMode turbotunnel.QueueOverflowMode) (net.PacketConn, net.Addr, error) { switch r.ResolverType { case ResolverTypeUDP: addr, err := net.ResolveUDPAddr("udp", r.ResolverAddr) From 4dcdee0310fe7729500c3c9be5b541dfbe93a9a4 Mon Sep 17 00:00:00 2001 From: crazydi4mond <255249920+crazydi4mond@users.noreply.github.com> Date: Sun, 12 Apr 2026 00:36:37 +0200 Subject: [PATCH 03/13] fix(multi-resolver): isolate single-entry transport errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A read or write error on any one resolver entry previously propagated to DNSPacketConn via recvChan, tearing down the entire tunnel session — the opposite of what multi-resolver is meant to deliver. In practice only DoT triggered the cascade (UDP/DoH workers retry silently), but a mixed setup with one flaky DoT would reconnect in a loop. - Reader goroutines now mark their entry dead on transport error and exit, instead of pushing error-bearing results onto recvChan. - resolverEntry gains a sticky atomic dead flag; selection helpers and probe targeting skip dead entries; recomputeState short-circuits on them so the decay logic cannot flip Down back to Unknown. - MultiResolver gains an aliveCount/allDead pair; the last dying reader closes allDead, causing ReadFrom/WriteTo to return a terminal "all resolvers are down" error. This is the only condition that should cascade to tunnel reconnect. - WriteTo retries with the next alive entry on a writePacket failure, marking the failed entry dead. Error wrapping handles the empty case without a %w nil. - ReadFrom drains any buffered packet from a deceased reader before surfacing the terminal error, avoiding packet loss at teardown. - markDead holds mu during the transition and returns true only on the alive-to-dead edge, so concurrent observers (reader + writer seeing the same Close) cannot double-decrement aliveCount. Unit tests in client/multi_resolver_test.go exercise both symptoms with a fake net.PacketConn: - TestMultiResolver_DeadEntryDoesNotBreakReadFrom — ReadFrom must return bytes from the working entry after a peer errors. - TestMultiResolver_FailedEntryExcludedFromSelection — selectPrimary must stop returning the failed entry after its reader exits. --- client/multi_resolver.go | 186 +++++++++++++++++++++++---- client/multi_resolver_test.go | 229 ++++++++++++++++++++++++++++++++++ 2 files changed, 389 insertions(+), 26 deletions(-) create mode 100644 client/multi_resolver_test.go diff --git a/client/multi_resolver.go b/client/multi_resolver.go index c46f791..d1fea54 100644 --- a/client/multi_resolver.go +++ b/client/multi_resolver.go @@ -66,6 +66,13 @@ type resolverEntry struct { invalidCount atomic.Int64 timeoutCount atomic.Int64 + // dead is set when the entry's transport has reported an unrecoverable + // error and its reader goroutine has exited. A dead entry is permanently + // excluded from selection and never transitions back — the only way to + // recover is to rebuild the entire MultiResolver (which happens on a + // full tunnel reconnect). + dead atomic.Bool + mu sync.Mutex pending map[uint16]time.Time lastWrite time.Time @@ -74,6 +81,31 @@ type resolverEntry struct { state ResolverState } +// markDead transitions the entry to a permanent Down state. It is idempotent +// and returns true only when this call was the one that transitioned the +// entry — callers rely on this to avoid double-counting the entry in +// MultiResolver.aliveCount when multiple goroutines detect the same transport +// failure concurrently. +// +// The dead flag and the state field are set together under mu so that +// recomputeState (also holding mu) cannot observe an inconsistent snapshot +// where dead is set but state has not yet been transitioned to Down. +func (e *resolverEntry) markDead() bool { + e.mu.Lock() + defer e.mu.Unlock() + if e.dead.Load() { + return false + } + e.dead.Store(true) + e.state = ResolverStateDown + return true +} + +// isDead reports whether markDead has been called on this entry. +func (e *resolverEntry) isDead() bool { + return e.dead.Load() +} + func (e *resolverEntry) writePacket(b []byte) (int, error) { now := time.Now() e.trackOutgoingID(b, now) @@ -156,8 +188,17 @@ func (e *resolverEntry) expirePending(now time.Time) { } func (e *resolverEntry) recomputeState(now time.Time) { + // Dead is sticky — never recompute it back to any other state. We + // re-check under the lock after acquiring it, in case markDead raced + // in between the outer check and acquiring the lock. + if e.isDead() { + return + } e.mu.Lock() defer e.mu.Unlock() + if e.dead.Load() { + return + } timeouts := e.timeoutCount.Load() invalid := e.invalidCount.Load() @@ -238,6 +279,15 @@ type MultiResolver struct { recvChan chan multiReadResult closed chan struct{} closeOnce sync.Once + + // aliveCount tracks how many entries still have a live reader + // goroutine. When it reaches zero, allDead is closed, causing pending + // ReadFrom/WriteTo calls to return an "all resolvers down" error, + // which is the only condition that should propagate a transport + // error to the upper tunnel session. + aliveCount atomic.Int32 + allDead chan struct{} + allDeadOnce sync.Once } // NewMultiResolver creates a MultiResolver from a slice of Resolver configs. @@ -269,25 +319,57 @@ func NewMultiResolver(resolvers []Resolver, mode SelectionMode, queueSize int, o mode: mode, recvChan: make(chan multiReadResult, len(entries)*4), closed: make(chan struct{}), + allDead: make(chan struct{}), } - for _, e := range entries { + mr.aliveCount.Store(int32(len(entries))) + mr.startReaders() + go mr.healthWorker() + return mr, nil +} + +// entryDied is called whenever a transport error is observed for an entry, +// either from its reader goroutine or from a failed write. It marks the entry +// dead and, if this was the transition from alive to dead (as opposed to a +// second observer of the same failure), decrements aliveCount. When the last +// alive entry dies, allDead is closed so pending ReadFrom/WriteTo callers can +// unblock with a terminal error. +func (mr *MultiResolver) entryDied(entry *resolverEntry, err error) { + if !entry.markDead() { + return + } + log.Warnf("multi-resolver: entry %s transport error: %v; marking down", entry.name, err) + if mr.aliveCount.Add(-1) == 0 { + mr.allDeadOnce.Do(func() { close(mr.allDead) }) + } +} + +// startReaders launches one reader goroutine per entry. Each goroutine reads +// packets from its entry's transport and pushes the result onto recvChan. +// Extracted so tests can construct MultiResolver with synthetic entries. +// +// On a transport error, the reader calls entryDied to mark the entry dead and +// signal allDead when the last entry dies. It never pushes error-bearing +// results onto recvChan: doing so would surface a single resolver's failure +// to DNSPacketConn.recvLoop, which would tear down the entire tunnel session +// and defeat the point of having multiple resolvers. +func (mr *MultiResolver) startReaders() { + for _, e := range mr.entries { entry := e go func() { for { res := entry.readPacket() + if res.err != nil { + mr.entryDied(entry, res.err) + return + } select { case mr.recvChan <- res: case <-mr.closed: return } - if res.err != nil { - return - } } }() } - go mr.healthWorker() - return mr, nil } func (mr *MultiResolver) healthWorker() { @@ -350,32 +432,66 @@ func isRateLimitedResponse(resp dns.Message) bool { } // ReadFrom receives a packet from whichever resolver responds first. +// It only returns an error when the MultiResolver has been closed or every +// entry has died; a single resolver's transport error is isolated at the +// reader goroutine and does not propagate here. func (mr *MultiResolver) ReadFrom(b []byte) (n int, addr net.Addr, err error) { select { case <-mr.closed: return 0, nil, net.ErrClosed case res := <-mr.recvChan: - if res.err != nil { - return 0, res.addr, res.err - } n = copy(b, res.buf[:res.n]) return n, turbotunnel.DummyAddr{}, nil + case <-mr.allDead: + // Drain any packet that was buffered by a reader before it + // died, so packets already delivered by the transport are not + // discarded in favour of the terminal error. No more readers + // are pushing (allDead is closed only after every reader has + // exited), so a non-blocking read here races only with the + // mr.closed case above, which is handled on the next call. + select { + case res := <-mr.recvChan: + n = copy(b, res.buf[:res.n]) + return n, turbotunnel.DummyAddr{}, nil + default: + return 0, nil, fmt.Errorf("multi-resolver: all resolvers are down") + } } } // WriteTo sends b to the selected primary resolver and may duplicate b to one -// unhealthy resolver as a probe to detect recovery. +// unhealthy resolver as a probe to detect recovery. If the primary's write +// fails, the entry is marked dead and WriteTo retries with the next alive +// entry. An error is returned only when the MultiResolver is closed or every +// entry has been marked dead. func (mr *MultiResolver) WriteTo(b []byte, _ net.Addr) (n int, err error) { select { case <-mr.closed: return 0, net.ErrClosed + case <-mr.allDead: + return 0, fmt.Errorf("multi-resolver: all resolvers are down") default: } - primary := mr.selectPrimary() - n, err = primary.writePacket(b) - if err != nil { - return n, err + // Try entries until one accepts the write or all alive entries have + // been exhausted. Each write error marks the entry dead. + var primary *resolverEntry + for attempts := 0; attempts < len(mr.entries); attempts++ { + primary = mr.selectPrimary() + if primary == nil { + break + } + n, err = primary.writePacket(b) + if err == nil { + break + } + mr.entryDied(primary, err) + } + switch { + case err != nil: + return 0, fmt.Errorf("multi-resolver: all write attempts failed: %w", err) + case primary == nil: + return 0, fmt.Errorf("multi-resolver: no alive resolver for write") } if probe := mr.selectProbeTarget(primary); probe != nil { @@ -385,17 +501,17 @@ func (mr *MultiResolver) WriteTo(b []byte, _ net.Addr) (n int, err error) { return n, nil } +// selectPrimary returns the entry that should receive the next outgoing +// query, or nil if every entry has been marked dead. The caller must handle +// nil (e.g., return an "all resolvers down" error). func (mr *MultiResolver) selectPrimary() *resolverEntry { if mr.mode == SelectionRoundRobin { - if e := mr.selectRoundRobinHealthy(); e != nil { - return e - } - return mr.entries[0] + return mr.selectRoundRobinHealthy() } if e := mr.selectBestScore(); e != nil { return e } - return mr.entries[0] + return mr.selectRoundRobinHealthy() } func (mr *MultiResolver) selectRoundRobinHealthy() *resolverEntry { @@ -407,28 +523,43 @@ func (mr *MultiResolver) selectRoundRobinHealthy() *resolverEntry { } start := mr.rrIndex + // First pass: prefer Healthy or Unknown, skipping dead entries. for i := 0; i < len(mr.entries); i++ { idx := (start + i) % len(mr.entries) + if mr.entries[idx].isDead() { + continue + } state := mr.entries[idx].stateSnapshot() if state == ResolverStateHealthy || state == ResolverStateUnknown { mr.rrIndex = (idx + 1) % len(mr.entries) return mr.entries[idx] } } - idx := start % len(mr.entries) - mr.rrIndex = (idx + 1) % len(mr.entries) - return mr.entries[idx] + // Second pass: accept any non-dead entry even if RateLimited/Down. + for i := 0; i < len(mr.entries); i++ { + idx := (start + i) % len(mr.entries) + if mr.entries[idx].isDead() { + continue + } + mr.rrIndex = (idx + 1) % len(mr.entries) + return mr.entries[idx] + } + // Every entry is dead. + return nil } func (mr *MultiResolver) selectBestScore() *resolverEntry { if len(mr.entries) == 0 { return nil } - best := mr.entries[0] - bestScore := resolverScore(best) - for _, e := range mr.entries[1:] { + var best *resolverEntry + var bestScore int64 + for _, e := range mr.entries { + if e.isDead() { + continue + } s := resolverScore(e) - if s > bestScore { + if best == nil || s > bestScore { best = e bestScore = s } @@ -462,6 +593,9 @@ func (mr *MultiResolver) selectProbeTarget(primary *resolverEntry) *resolverEntr if e == primary { continue } + if e.isDead() { + continue + } state := e.stateSnapshot() if state == ResolverStateHealthy { continue diff --git a/client/multi_resolver_test.go b/client/multi_resolver_test.go new file mode 100644 index 0000000..ec31411 --- /dev/null +++ b/client/multi_resolver_test.go @@ -0,0 +1,229 @@ +package client + +import ( + "bytes" + "net" + "sync" + "testing" + "time" + + "github.com/net2share/vaydns/turbotunnel" +) + +// fakePacketConn is a controllable net.PacketConn for MultiResolver tests. +// Pushed responses are consumed in order by successive ReadFrom calls; each +// response can be either data bytes or an error. +type fakePacketConn struct { + name string + readCh chan fakeReadResp + closed chan struct{} + once sync.Once +} + +type fakeReadResp struct { + data []byte + err error +} + +func newFakePacketConn(name string) *fakePacketConn { + return &fakePacketConn{ + name: name, + readCh: make(chan fakeReadResp, 16), + closed: make(chan struct{}), + } +} + +func (f *fakePacketConn) pushData(data []byte) { + cp := make([]byte, len(data)) + copy(cp, data) + f.readCh <- fakeReadResp{data: cp} +} + +func (f *fakePacketConn) pushError(err error) { + f.readCh <- fakeReadResp{err: err} +} + +func (f *fakePacketConn) ReadFrom(p []byte) (int, net.Addr, error) { + select { + case r, ok := <-f.readCh: + if !ok { + return 0, nil, net.ErrClosed + } + if r.err != nil { + return 0, nil, r.err + } + return copy(p, r.data), turbotunnel.DummyAddr{}, nil + case <-f.closed: + return 0, nil, net.ErrClosed + } +} + +func (f *fakePacketConn) WriteTo(p []byte, _ net.Addr) (int, error) { + return len(p), nil +} + +func (f *fakePacketConn) Close() error { + f.once.Do(func() { close(f.closed) }) + return nil +} + +func (f *fakePacketConn) LocalAddr() net.Addr { return turbotunnel.DummyAddr{} } +func (f *fakePacketConn) SetDeadline(time.Time) error { return nil } +func (f *fakePacketConn) SetReadDeadline(time.Time) error { return nil } +func (f *fakePacketConn) SetWriteDeadline(time.Time) error { return nil } + +// newFakeEntry builds a resolverEntry around a fakePacketConn, suitable for +// injection into a hand-constructed MultiResolver in tests. +func newFakeEntry(name string, conn *fakePacketConn) *resolverEntry { + return &resolverEntry{ + name: name, + addr: turbotunnel.DummyAddr{}, + conn: conn, + pending: make(map[uint16]time.Time), + state: ResolverStateUnknown, + } +} + +// newTestMultiResolver mirrors what NewMultiResolver does after +// GetResolverConnection returns, but with pre-built synthetic entries. It does +// not spawn the healthWorker — tests don't need it, and omitting it keeps them +// hermetic with respect to timing. +func newTestMultiResolver(entries []*resolverEntry, mode SelectionMode) *MultiResolver { + mr := &MultiResolver{ + entries: entries, + mode: mode, + recvChan: make(chan multiReadResult, len(entries)*4), + closed: make(chan struct{}), + allDead: make(chan struct{}), + } + mr.aliveCount.Store(int32(len(entries))) + mr.startReaders() + return mr +} + +// TestMultiResolver_DeadEntryDoesNotBreakReadFrom exercises C1: when one +// resolver entry's transport returns a fatal read error, MultiResolver.ReadFrom +// must continue to deliver packets from healthy entries instead of propagating +// that error to the upper DNSPacketConn layer (which would tear down the whole +// tunnel session). +// +// Pre-fix expectation: this test FAILS — the reader goroutine pushes the error +// onto recvChan, ReadFrom returns it, and the test sees net.ErrClosed instead +// of the bytes from the working entry. +// +// Post-fix expectation: this test PASSES — the reader goroutine handles the +// error internally (e.g. marks the entry Down and exits), leaving recvChan to +// deliver only real response bytes from the surviving entries. +func TestMultiResolver_DeadEntryDoesNotBreakReadFrom(t *testing.T) { + failing := newFakePacketConn("failing") + working := newFakePacketConn("working") + + entries := []*resolverEntry{ + newFakeEntry("failing", failing), + newFakeEntry("working", working), + } + mr := newTestMultiResolver(entries, SelectionRoundRobin) + defer mr.Close() + + // Step 1: make the failing entry's ReadFrom return a fatal error. + // The reader goroutine will pick this up immediately. + failing.pushError(net.ErrClosed) + + // Give the reader goroutine time to handle the error. Under the buggy + // implementation it pushes an error-bearing multiReadResult onto + // recvChan and exits; under a correct implementation it would quietly + // mark the entry down and exit without poisoning recvChan. + time.Sleep(100 * time.Millisecond) + + // Step 2: push a valid response to the working entry. This lands on + // recvChan strictly AFTER any push from the failing entry, so the + // ordering is deterministic. + wantResponse := []byte("valid-response-bytes-from-working-resolver") + working.pushData(wantResponse) + + // Step 3: ReadFrom must return the bytes from the working entry, not + // the error from the failing one. + type readResult struct { + n int + err error + buf []byte + } + done := make(chan readResult, 1) + go func() { + buf := make([]byte, 4096) + n, _, err := mr.ReadFrom(buf) + done <- readResult{n: n, err: err, buf: buf} + }() + + select { + case r := <-done: + if r.err != nil { + t.Fatalf("MultiResolver.ReadFrom returned error %v; expected it to ignore the failed entry and return the bytes from the working entry. This is the C1 bug: a single resolver's read error tears down the tunnel.", r.err) + } + if !bytes.Equal(r.buf[:r.n], wantResponse) { + t.Fatalf("MultiResolver.ReadFrom returned wrong bytes.\n got: %x\n want: %x", r.buf[:r.n], wantResponse) + } + case <-time.After(3 * time.Second): + t.Fatal("MultiResolver.ReadFrom timed out; expected it to return bytes from the working entry within 3s") + } +} + +// TestMultiResolver_FailedEntryExcludedFromSelection exercises the selection +// side of C1: after a reader goroutine has processed a fatal read error and +// exited, the entry must no longer be returned by selectPrimary. Otherwise +// the round-robin scheduler will keep sending queries to a resolver whose +// reader goroutine is gone (so responses will never come back), defeating the +// health state machine. +// +// Pre-fix expectation: this test FAILS — the reader goroutine exits without +// updating the entry's state, so it remains ResolverStateUnknown, which +// selectRoundRobinHealthy treats as an acceptable target. The "failing" entry +// is returned roughly half the time. +// +// Post-fix expectation: this test PASSES — the reader goroutine transitions +// the entry to ResolverStateDown before exiting, and selectPrimary skips it. +func TestMultiResolver_FailedEntryExcludedFromSelection(t *testing.T) { + failing := newFakePacketConn("failing") + working := newFakePacketConn("working") + + entries := []*resolverEntry{ + newFakeEntry("failing", failing), + newFakeEntry("working", working), + } + mr := newTestMultiResolver(entries, SelectionRoundRobin) + defer mr.Close() + + // Kill the failing entry's reader; leave working blocked in ReadFrom. + failing.pushError(net.ErrClosed) + + // Let the reader goroutine process the error and (ideally) mark the + // entry down before it exits. + time.Sleep(100 * time.Millisecond) + + // Drain any error result that the buggy reader may have pushed onto + // recvChan. A correct reader wouldn't push anything here, so this + // drain is a no-op post-fix. Without it, the buggy pre-fix state + // machine would leave an error lingering, which isn't what this + // specific assertion is about. +drain: + for { + select { + case <-mr.recvChan: + default: + break drain + } + } + + // Call selectPrimary repeatedly. It must never return the failing + // entry — otherwise real queries will be sent to a resolver whose + // reader is gone, so responses will never come back. + for i := range 10 { + selected := mr.selectPrimary() + if selected == nil { + t.Fatalf("iter %d: selectPrimary returned nil", i) + } + if selected.name == "failing" { + t.Fatalf("iter %d: selectPrimary returned the failed entry %q; expected it to be excluded after its reader exited on error. State is %s.", i, selected.name, selected.stateSnapshot()) + } + } +} From 922faf0ff6c706d6330df9eb824d8ae8071c4afa Mon Sep 17 00:00:00 2001 From: crazydi4mond <255249920+crazydi4mond@users.noreply.github.com> Date: Sun, 12 Apr 2026 01:06:27 +0200 Subject: [PATCH 04/13] test(e2e): add multi-resolver integration tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Exercise the multi-resolver client path end-to-end with three Docker-based scenarios covering the happy path, a mid-flight resolver kill, and a forging resolver: - multi-resolver: two healthy CoreDNS forwarders, both serving the tunnel domain; verifies the client works with -udp flag repeated. - multi-resolver-runtime-failure: two healthy resolvers, then docker kill one mid-flight; asserts HTTP traffic keeps flowing through the survivor AND that no new tunnel sessions are created — proving the single-entry failure is isolated, not escalated to a full session reconnect. - multi-resolver-forge: one healthy resolver plus one CoreDNS instance using the template plugin to inject NXDOMAIN for every query; asserts the tunnel still delivers while the UDP worker's forged-response filter absorbs the injections, and sanity-checks the forging path by grepping dns-forge's own logs for NXDOMAIN to rule out a round-robin false pass. All three tests use explicit -idle-timeout 10s -keepalive 2s on both client and server (matching recovery/transport-recovery) so they stay anchored to a production-like timing profile and won't drift when the branch's defaults are later reconciled with main. The full e2e suite runner (run-test.sh) is updated to include the three new tests after transport-recovery. --- e2e/multi-resolver-forge/Corefile.forge | 6 ++ e2e/multi-resolver-forge/docker-compose.yml | 82 +++++++++++++++++++ e2e/multi-resolver-forge/run.sh | 59 +++++++++++++ .../docker-compose.yml | 82 +++++++++++++++++++ e2e/multi-resolver-runtime-failure/run.sh | 82 +++++++++++++++++++ e2e/multi-resolver/docker-compose.yml | 82 +++++++++++++++++++ e2e/multi-resolver/run.sh | 29 +++++++ e2e/run-test.sh | 2 +- 8 files changed, 423 insertions(+), 1 deletion(-) create mode 100644 e2e/multi-resolver-forge/Corefile.forge create mode 100644 e2e/multi-resolver-forge/docker-compose.yml create mode 100755 e2e/multi-resolver-forge/run.sh create mode 100644 e2e/multi-resolver-runtime-failure/docker-compose.yml create mode 100755 e2e/multi-resolver-runtime-failure/run.sh create mode 100644 e2e/multi-resolver/docker-compose.yml create mode 100755 e2e/multi-resolver/run.sh diff --git a/e2e/multi-resolver-forge/Corefile.forge b/e2e/multi-resolver-forge/Corefile.forge new file mode 100644 index 0000000..224cbc9 --- /dev/null +++ b/e2e/multi-resolver-forge/Corefile.forge @@ -0,0 +1,6 @@ +. { + template IN ANY . { + rcode NXDOMAIN + } + log +} diff --git a/e2e/multi-resolver-forge/docker-compose.yml b/e2e/multi-resolver-forge/docker-compose.yml new file mode 100644 index 0000000..b5be9f2 --- /dev/null +++ b/e2e/multi-resolver-forge/docker-compose.yml @@ -0,0 +1,82 @@ +networks: + dns-net: + ipam: + config: + - subnet: 172.28.0.0/24 + backend-net: + +volumes: + keys: + +services: + keygen: + build: + context: ../.. + dockerfile: Dockerfile + volumes: + - keys:/keys + command: > + sh -c "vaydns-server -gen-key -privkey-file /keys/server.key -pubkey-file /keys/server.pub" + + dns-good: + image: coredns/coredns + networks: + dns-net: + ipv4_address: 172.28.0.10 + volumes: + - ../Corefile:/Corefile + command: ["-conf", "/Corefile"] + + dns-forge: + image: coredns/coredns + networks: + dns-net: + ipv4_address: 172.28.0.11 + volumes: + - ./Corefile.forge:/Corefile + command: ["-conf", "/Corefile"] + + backend: + image: nginx:alpine + networks: + - backend-net + + server: + build: + context: ../.. + dockerfile: Dockerfile + networks: + dns-net: + ipv4_address: 172.28.0.20 + backend-net: + volumes: + - keys:/keys + command: > + vaydns-server -udp :53 -privkey-file /keys/server.key + -domain t.example.com -upstream backend:80 + -idle-timeout 10s -keepalive 2s + depends_on: + keygen: + condition: service_completed_successfully + dns-good: + condition: service_started + dns-forge: + condition: service_started + + client: + build: + context: ../.. + dockerfile: Dockerfile + networks: + - dns-net + volumes: + - keys:/keys + command: > + vaydns-client -udp 172.28.0.10:53 -udp 172.28.0.11:53 + -pubkey-file /keys/server.pub + -domain t.example.com -listen 0.0.0.0:7000 + -idle-timeout 10s -keepalive 2s -session-check-interval 500ms + -reconnect-min 1s -reconnect-max 5s -log-level info + depends_on: + server: + condition: service_started diff --git a/e2e/multi-resolver-forge/run.sh b/e2e/multi-resolver-forge/run.sh new file mode 100755 index 0000000..418fa3c --- /dev/null +++ b/e2e/multi-resolver-forge/run.sh @@ -0,0 +1,59 @@ +#!/usr/bin/env bash +# Test: one resolver returns forged responses while another works. +# dns-forge always replies with NXDOMAIN (CoreDNS template plugin), simulating +# a censor or broken resolver injecting fake responses. dns-good behaves +# normally. The tunnel must work through dns-good, with the per-query UDP +# worker's forged-response filter absorbing dns-forge's NXDOMAINs and +# MultiResolver's health state machine eventually routing around it. +set -euo pipefail +cd "$(dirname "$0")" + +cleanup() { docker compose down -v 2>/dev/null; } +trap cleanup EXIT + +fetch() { + docker compose exec -T client wget -q -O- http://localhost:7000 2>/dev/null | grep -q "Welcome to nginx" +} + +echo "--- Building and starting services ---" +docker compose up -d --build + +echo "--- Waiting for tunnel through dns-good while dns-forge injects NXDOMAINs (up to 60s) ---" +ok_count=0 +for i in $(seq 1 60); do + if fetch; then + ok_count=$((ok_count + 1)) + # Require two consecutive successes so one lucky query doesn't pass + # the test while the forging resolver is still in rotation. + if [ "$ok_count" -ge 2 ]; then + echo "" + # Sanity check: make sure dns-forge actually saw queries, so we + # know the tunnel was exercising the forging code path and didn't + # only hit dns-good by chance. CoreDNS's log plugin may buffer + # output briefly, so give it a moment to flush before grepping. + sleep 2 + forge_logs=$(docker compose logs dns-forge 2>&1) + if ! grep -q 'NXDOMAIN' <<<"$forge_logs"; then + echo "--- dns-forge never served NXDOMAIN; the forging code path may not have been exercised ---" + echo "$forge_logs" + echo "=== FAIL (forging path not exercised) ===" + exit 1 + fi + nxdomain_count=$(grep -c 'NXDOMAIN' <<<"$forge_logs" || true) + echo "--- dns-forge served $nxdomain_count NXDOMAIN responses; forging path exercised ---" + echo "--- Tunnel delivers consistent responses despite forged NXDOMAINs ---" + echo "=== PASS ===" + exit 0 + fi + else + ok_count=0 + fi + printf "." + sleep 1 +done + +echo "" +echo "--- Tunnel did not come up through dns-good ---" +docker compose logs client server dns-good dns-forge +echo "=== FAIL ===" +exit 1 diff --git a/e2e/multi-resolver-runtime-failure/docker-compose.yml b/e2e/multi-resolver-runtime-failure/docker-compose.yml new file mode 100644 index 0000000..d456182 --- /dev/null +++ b/e2e/multi-resolver-runtime-failure/docker-compose.yml @@ -0,0 +1,82 @@ +networks: + dns-net: + ipam: + config: + - subnet: 172.28.0.0/24 + backend-net: + +volumes: + keys: + +services: + keygen: + build: + context: ../.. + dockerfile: Dockerfile + volumes: + - keys:/keys + command: > + sh -c "vaydns-server -gen-key -privkey-file /keys/server.key -pubkey-file /keys/server.pub" + + dns1: + image: coredns/coredns + networks: + dns-net: + ipv4_address: 172.28.0.10 + volumes: + - ../Corefile:/Corefile + command: ["-conf", "/Corefile"] + + dns2: + image: coredns/coredns + networks: + dns-net: + ipv4_address: 172.28.0.11 + volumes: + - ../Corefile:/Corefile + command: ["-conf", "/Corefile"] + + backend: + image: nginx:alpine + networks: + - backend-net + + server: + build: + context: ../.. + dockerfile: Dockerfile + networks: + dns-net: + ipv4_address: 172.28.0.20 + backend-net: + volumes: + - keys:/keys + command: > + vaydns-server -udp :53 -privkey-file /keys/server.key + -domain t.example.com -upstream backend:80 + -idle-timeout 10s -keepalive 2s + depends_on: + keygen: + condition: service_completed_successfully + dns1: + condition: service_started + dns2: + condition: service_started + + client: + build: + context: ../.. + dockerfile: Dockerfile + networks: + - dns-net + volumes: + - keys:/keys + command: > + vaydns-client -udp 172.28.0.10:53 -udp 172.28.0.11:53 + -pubkey-file /keys/server.pub + -domain t.example.com -listen 0.0.0.0:7000 + -idle-timeout 10s -keepalive 2s -session-check-interval 500ms + -reconnect-min 1s -reconnect-max 5s -log-level info + depends_on: + server: + condition: service_started diff --git a/e2e/multi-resolver-runtime-failure/run.sh b/e2e/multi-resolver-runtime-failure/run.sh new file mode 100755 index 0000000..a863b79 --- /dev/null +++ b/e2e/multi-resolver-runtime-failure/run.sh @@ -0,0 +1,82 @@ +#!/usr/bin/env bash +# Test: one of several UDP resolvers dies mid-flight. +# Start the tunnel with two working DNS resolvers, verify HTTP traffic works, +# kill one resolver, and verify traffic continues through the remaining one. +# +# A failing UDP resolver does not surface an error to MultiResolver (the +# per-query worker retries silently), so this test validates that the health +# state machine + round-robin selection routes around it without tearing +# down the tunnel session. +set -euo pipefail +cd "$(dirname "$0")" + +cleanup() { docker compose down -v 2>/dev/null; } +trap cleanup EXIT + +fetch() { + docker compose exec -T client wget -q -O- http://localhost:7000 2>/dev/null | grep -q "Welcome to nginx" +} + +echo "--- Building and starting services ---" +docker compose up -d --build + +echo "--- Waiting for initial tunnel (up to 30s) ---" +for i in $(seq 1 30); do + if fetch; then + echo "" + echo "--- Initial tunnel is up ---" + break + fi + if [ "$i" -eq 30 ]; then + echo "" + docker compose logs client server dns1 dns2 + echo "=== FAIL (initial tunnel not ready) ===" + exit 1 + fi + printf "." + sleep 1 +done + +# Snapshot the session id before the kill, to detect any reconnect later. +pre_kill_sessions=$(docker compose logs client 2>&1 | grep -c 'session .* ready' || true) + +echo "--- Killing dns1 (half the queries will start dropping) ---" +docker compose kill dns1 + +# Give the client a moment to notice and start routing around dns1. +sleep 3 + +echo "--- Verifying tunnel still works through dns2 (up to 45s) ---" +ok_count=0 +for i in $(seq 1 45); do + if fetch; then + ok_count=$((ok_count + 1)) + # Require two consecutive successes so we don't declare victory on + # a lucky query that happened to go to dns2. + if [ "$ok_count" -ge 2 ]; then + post_kill_sessions=$(docker compose logs client 2>&1 | grep -c 'session .* ready' || true) + new_sessions=$((post_kill_sessions - pre_kill_sessions)) + if [ "$new_sessions" -gt 0 ]; then + echo "" + echo "--- Tunnel recovered but triggered $new_sessions new session(s) — resolver failure should be isolated without a full reconnect ---" + docker compose logs client | tail -30 + echo "=== FAIL (session was rebuilt instead of isolated) ===" + exit 1 + fi + echo "" + echo "--- Tunnel survived with 0 new sessions (single-entry failure isolated) ---" + echo "=== PASS ===" + exit 0 + fi + else + ok_count=0 + fi + printf "." + sleep 1 +done + +echo "" +echo "--- Tunnel did not survive dns1 kill ---" +docker compose logs client server dns1 dns2 +echo "=== FAIL ===" +exit 1 diff --git a/e2e/multi-resolver/docker-compose.yml b/e2e/multi-resolver/docker-compose.yml new file mode 100644 index 0000000..36f9222 --- /dev/null +++ b/e2e/multi-resolver/docker-compose.yml @@ -0,0 +1,82 @@ +networks: + dns-net: + ipam: + config: + - subnet: 172.28.0.0/24 + backend-net: + +volumes: + keys: + +services: + keygen: + build: + context: ../.. + dockerfile: Dockerfile + volumes: + - keys:/keys + command: > + sh -c "vaydns-server -gen-key -privkey-file /keys/server.key -pubkey-file /keys/server.pub" + + dns1: + image: coredns/coredns + networks: + dns-net: + ipv4_address: 172.28.0.10 + volumes: + - ../Corefile:/Corefile + command: ["-conf", "/Corefile"] + + dns2: + image: coredns/coredns + networks: + dns-net: + ipv4_address: 172.28.0.11 + volumes: + - ../Corefile:/Corefile + command: ["-conf", "/Corefile"] + + backend: + image: nginx:alpine + networks: + - backend-net + + server: + build: + context: ../.. + dockerfile: Dockerfile + networks: + dns-net: + ipv4_address: 172.28.0.20 + backend-net: + volumes: + - keys:/keys + command: > + vaydns-server -udp :53 -privkey-file /keys/server.key + -domain t.example.com -upstream backend:80 + -idle-timeout 10s -keepalive 2s + depends_on: + keygen: + condition: service_completed_successfully + dns1: + condition: service_started + dns2: + condition: service_started + + client: + build: + context: ../.. + dockerfile: Dockerfile + networks: + - dns-net + volumes: + - keys:/keys + command: > + vaydns-client -udp 172.28.0.10:53 -udp 172.28.0.11:53 + -pubkey-file /keys/server.pub + -domain t.example.com -listen 0.0.0.0:7000 + -idle-timeout 10s -keepalive 2s -session-check-interval 500ms + -reconnect-min 1s -reconnect-max 5s + depends_on: + server: + condition: service_started diff --git a/e2e/multi-resolver/run.sh b/e2e/multi-resolver/run.sh new file mode 100755 index 0000000..c64921d --- /dev/null +++ b/e2e/multi-resolver/run.sh @@ -0,0 +1,29 @@ +#!/usr/bin/env bash +# Test: multi-resolver smoke test. +# Verifies that vaydns-client works when configured with two UDP resolvers, +# and that an HTTP request through the tunnel succeeds. +set -euo pipefail +cd "$(dirname "$0")" + +cleanup() { docker compose down -v 2>/dev/null; } +trap cleanup EXIT + +echo "--- Building and starting services ---" +docker compose up -d --build + +echo "--- Waiting for tunnel (up to 30s) ---" +for i in $(seq 1 30); do + if docker compose exec -T client wget -q -O- http://localhost:7000 2>/dev/null | grep -q "Welcome to nginx"; then + echo "" + echo "=== PASS ===" + exit 0 + fi + printf "." + sleep 1 +done + +echo "" +echo "--- Tunnel did not come up. Dumping logs ---" +docker compose logs client server dns1 dns2 +echo "=== FAIL ===" +exit 1 diff --git a/e2e/run-test.sh b/e2e/run-test.sh index b07c0b7..9784ae1 100755 --- a/e2e/run-test.sh +++ b/e2e/run-test.sh @@ -20,7 +20,7 @@ for rt in txt cname a aaaa mx ns srv; do fi done -for test_dir in socks-download recovery transport-recovery; do +for test_dir in socks-download recovery transport-recovery multi-resolver multi-resolver-runtime-failure multi-resolver-forge; do total=$((total + 1)) echo "" echo "========================================" From fcd228a2b98bcf371fa984b4735604ec3f31c2d8 Mon Sep 17 00:00:00 2001 From: crazydi4mond <255249920+crazydi4mond@users.noreply.github.com> Date: Sun, 12 Apr 2026 01:31:44 +0200 Subject: [PATCH 05/13] fix(client): apply uTLS fingerprint to DoT resolvers in multi mode The multi-resolver refactor split the single "resolver" variable into three per-type loops but only re-added the UTLSClientHelloID assignment to the DoH loop, silently dropping fingerprint camouflage for every DoT resolver. Pre-refactor main set this field unconditionally on the single configured resolver, so single-resolver DoT users on main got their fingerprint; multi-resolver DoT users on this branch do not. Restore the assignment in the DoT loop, mirroring the DoH loop's pattern. --- vaydns-client/main.go | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/vaydns-client/main.go b/vaydns-client/main.go index 0924f7e..a93d35f 100644 --- a/vaydns-client/main.go +++ b/vaydns-client/main.go @@ -238,10 +238,12 @@ Known TLS fingerprints for -utls are: resolvers = append(resolvers, resolver) } for _, dotAddr := range dotAddrs { - resolvers = append(resolvers, client.Resolver{ + resolver := client.Resolver{ ResolverType: client.ResolverTypeDOT, ResolverAddr: dotAddr, - }) + } + resolver.UTLSClientHelloID = utlsClientHelloID + resolvers = append(resolvers, resolver) } for _, udpAddr := range udpAddrs { resolver := client.Resolver{ From 6f35ceb1fa292d807410186321348ba53256c3de Mon Sep 17 00:00:00 2001 From: crazydi4mond <255249920+crazydi4mond@users.noreply.github.com> Date: Sun, 12 Apr 2026 01:43:48 +0200 Subject: [PATCH 06/13] fix(multi-resolver): move health counters under mu to close decay race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The decay logic in recomputeState read each counter with atomic Load, then wrote back Load-1 with atomic Store. Concurrent Adds from writePacket, evaluateIncoming, and expirePending (all running in different goroutines) could land between the Load and Store and be silently overwritten — classic lost-update on the invalid/timeout counters. timeoutCount.Store(0) in the valid-response path had the same shape. The Go race detector does not catch this: mixed atomic operations on the same field are considered synchronized, and the bug is a logic lost-update rather than an unsynchronized memory access. Move validCount, invalidCount, and timeoutCount from atomic.Int64 to plain int64 fields protected by the entry's existing mu, and consolidate the several small mu acquisitions in evaluateIncoming, expirePending, and writePacket into one per branch. recomputeState's decay becomes a simple "if count > 0 { count-- }" under the lock. dead and MultiResolver.aliveCount remain atomic: they are read from selection hot paths that should not acquire e.mu on every iteration. resolverScore now acquires e.mu instead of calling stateSnapshot() and reading counters separately — side benefit: eliminates a minor pre-existing inconsistency where the state and the counter values could come from different snapshots. --- client/multi_resolver.go | 106 +++++++++++++++++++++------------------ 1 file changed, 56 insertions(+), 50 deletions(-) diff --git a/client/multi_resolver.go b/client/multi_resolver.go index d1fea54..1b34535 100644 --- a/client/multi_resolver.go +++ b/client/multi_resolver.go @@ -62,23 +62,28 @@ type resolverEntry struct { addr net.Addr conn net.PacketConn - validCount atomic.Int64 - invalidCount atomic.Int64 - timeoutCount atomic.Int64 - // dead is set when the entry's transport has reported an unrecoverable // error and its reader goroutine has exited. A dead entry is permanently // excluded from selection and never transitions back — the only way to // recover is to rebuild the entire MultiResolver (which happens on a - // full tunnel reconnect). + // full tunnel reconnect). Kept as an atomic.Bool so selection helpers + // can read it without acquiring mu on the hot path. dead atomic.Bool - mu sync.Mutex - pending map[uint16]time.Time - lastWrite time.Time - lastValid time.Time - lastProbe time.Time - state ResolverState + // mu protects every field below. Counters are deliberately not atomic: + // the decay path in recomputeState would lose concurrent Adds if the + // counters were atomic and decremented via Load→Store, and splitting + // decay across a CAS loop is uglier than just holding mu for the + // short window each counter update requires. + mu sync.Mutex + validCount int64 + invalidCount int64 + timeoutCount int64 + pending map[uint16]time.Time + lastWrite time.Time + lastValid time.Time + lastProbe time.Time + state ResolverState } // markDead transitions the entry to a permanent Down state. It is idempotent @@ -110,11 +115,11 @@ func (e *resolverEntry) writePacket(b []byte) (int, error) { now := time.Now() e.trackOutgoingID(b, now) n, err := e.conn.WriteTo(b, e.addr) - if err != nil { - e.invalidCount.Add(1) - } e.mu.Lock() e.lastWrite = now + if err != nil { + e.invalidCount++ + } e.mu.Unlock() return n, err } @@ -140,50 +145,51 @@ func (e *resolverEntry) readPacket() multiReadResult { } func (e *resolverEntry) evaluateIncoming(packet []byte) { + now := time.Now() resp, err := dns.MessageFromWireFormat(packet) if err != nil { - e.invalidCount.Add(1) - e.recomputeState(time.Now()) + e.mu.Lock() + e.invalidCount++ + e.mu.Unlock() + e.recomputeState(now) return } - e.mu.Lock() - delete(e.pending, resp.ID) - e.mu.Unlock() - if isValidDNSResponse(resp) { - e.validCount.Add(1) - e.timeoutCount.Store(0) e.mu.Lock() - e.lastValid = time.Now() + delete(e.pending, resp.ID) + e.validCount++ + e.timeoutCount = 0 + e.lastValid = now e.state = ResolverStateHealthy e.mu.Unlock() return } - e.invalidCount.Add(1) + e.mu.Lock() + delete(e.pending, resp.ID) + e.invalidCount++ if isRateLimitedResponse(resp) { - e.mu.Lock() e.state = ResolverStateRateLimited - e.mu.Unlock() } - e.recomputeState(time.Now()) + e.mu.Unlock() + e.recomputeState(now) } func (e *resolverEntry) expirePending(now time.Time) { - expired := int64(0) e.mu.Lock() + expired := int64(0) for id, t := range e.pending { if now.Sub(t) >= pendingResponseTimeout { delete(e.pending, id) expired++ } } - e.mu.Unlock() if expired > 0 { - e.timeoutCount.Add(expired) - e.invalidCount.Add(expired) + e.timeoutCount += expired + e.invalidCount += expired } + e.mu.Unlock() e.recomputeState(now) } @@ -200,29 +206,25 @@ func (e *resolverEntry) recomputeState(now time.Time) { return } - timeouts := e.timeoutCount.Load() - invalid := e.invalidCount.Load() - valid := e.validCount.Load() - switch { - case timeouts >= downTimeoutThreshold: + case e.timeoutCount >= downTimeoutThreshold: e.state = ResolverStateDown - case invalid >= rateLimitThreshold && valid == 0: + case e.invalidCount >= rateLimitThreshold && e.validCount == 0: e.state = ResolverStateRateLimited - case valid > 0 && now.Sub(e.lastValid) <= 30*time.Second: + case e.validCount > 0 && now.Sub(e.lastValid) <= 30*time.Second: e.state = ResolverStateHealthy - case valid == 0: + case e.validCount == 0: e.state = ResolverStateUnknown default: e.state = ResolverStateUnknown } // Slow decay to avoid sticky penalties. - if invalid > 0 { - e.invalidCount.Store(invalid - 1) + if e.invalidCount > 0 { + e.invalidCount-- } - if timeouts > 0 { - e.timeoutCount.Store(timeouts - 1) + if e.timeoutCount > 0 { + e.timeoutCount-- } } @@ -250,9 +252,9 @@ func (e *resolverEntry) snapshot() ResolverStat { return ResolverStat{ Address: e.name, State: e.state, - ValidCount: e.validCount.Load(), - InvalidCount: e.invalidCount.Load(), - TimeoutCount: e.timeoutCount.Load(), + ValidCount: e.validCount, + InvalidCount: e.invalidCount, + TimeoutCount: e.timeoutCount, LastWrite: e.lastWrite, LastValid: e.lastValid, } @@ -568,8 +570,10 @@ func (mr *MultiResolver) selectBestScore() *resolverEntry { } func resolverScore(e *resolverEntry) int64 { - statePenalty := int64(0) - switch e.stateSnapshot() { + e.mu.Lock() + defer e.mu.Unlock() + var statePenalty int64 + switch e.state { case ResolverStateHealthy: statePenalty = 0 case ResolverStateUnknown: @@ -579,7 +583,7 @@ func resolverScore(e *resolverEntry) int64 { case ResolverStateDown: statePenalty = 30 } - return e.validCount.Load()*4 - e.invalidCount.Load()*2 - e.timeoutCount.Load()*3 - statePenalty + return e.validCount*4 - e.invalidCount*2 - e.timeoutCount*3 - statePenalty } func (mr *MultiResolver) selectProbeTarget(primary *resolverEntry) *resolverEntry { @@ -621,7 +625,9 @@ func (mr *MultiResolver) ResolverStats() []ResolverStat { func (mr *MultiResolver) ValidInvalidCounts() map[string][2]int64 { out := make(map[string][2]int64, len(mr.entries)) for _, e := range mr.entries { - out[e.name] = [2]int64{e.validCount.Load(), e.invalidCount.Load()} + e.mu.Lock() + out[e.name] = [2]int64{e.validCount, e.invalidCount} + e.mu.Unlock() } return out } From 04d4811a3f11293dbed6e079fdfa2d28f7de9faf Mon Sep 17 00:00:00 2001 From: crazydi4mond <255249920+crazydi4mond@users.noreply.github.com> Date: Sun, 12 Apr 2026 03:49:44 +0200 Subject: [PATCH 07/13] fix(multi-resolver): per-resolver forged response tracking with labeled logs Each resolver entry now owns a labeled ForgedStats instance so operators can see exactly which resolver is targeted by DNS injection. Milestone logs include the resolver address: forged DNS responses from 8.8.8.8:53: total=10 SERVFAIL=0 NXDOMAIN=10 other=0 In multi-resolver mode, the reader goroutine filters forged responses (QR=1, RCODE != NoError) at the MultiResolver layer and records them per-entry, preventing double-counting at the DNSPacketConn safety net. In single-resolver mode, ForgedStats is created by the Tunnel and shared between UDPPacketConn and DNSPacketConn, matching pre-PR behavior with the new labeled format. - ForgedStats gains a Label field; Record() includes "from