🏅 Sync 2026-10-25 22:22:42

This commit is contained in:
github-actions[bot] committed 2026-10-25 22:22:42 +08:00
1 parent 816dd89fe9
commit 53baf91238
134 files changed
+4664 -1277

No files matched your search

+1 -1
View File
@@ -6,7 +6,7 @@ include $(TOPDIR)/rules.mk
PKG_NAME:=mosdns
PKG_VERSION:=5.3.4
PKG_RELEASE:=22
PKG_RELEASE:=23
PKG_SOURCE:=$(PKG_NAME)-$(PKG_VERSION).tar.gz
PKG_SOURCE_URL:=https://codeload.github.com/IrineSistiana/mosdns/tar.gz/v$(PKG_VERSION)?
@@ -0,0 +1,21 @@
From 707d8f33174bfb403f3affe52d21be1e4c460cdd Mon Sep 17 00:00:00 2001
From: sbwml <admin@cooluc.com>
Date: Fri, 2 Oct 2026 01:01:08 +0800
Subject: [PATCH 1/5] fix(fallback): preserve respOpt with EDNS0 options from
winning branch
Signed-off-by: sbwml <admin@cooluc.com>
---
pkg/query_context/context.go | 1 +
1 file changed, 1 insertion(+)
--- a/pkg/query_context/context.go
+++ b/pkg/query_context/context.go
@@ -275,6 +275,7 @@ func (ctx *Context) CopyMetadataFrom(src
if src == nil {
return
}
+ ctx.respOpt = src.respOpt
ctx.upstreamOpt = src.upstreamOpt
ctx.UpstreamSelected = src.UpstreamSelected
if len(src.RuleHits) > 0 {
@@ -0,0 +1,183 @@
From 03e469715f297ea17be5090a6d64c4d2fbeb4e17 Mon Sep 17 00:00:00 2001
From: sbwml <admin@cooluc.com>
Date: Sat, 3 Oct 2026 23:40:31 +0800
Subject: [PATCH 2/5] fix(stats): exclude natural timeouts and upstream errors
from blocked statistics
---
plugin/executable/forward/forward.go | 45 ++++++++++++-------
plugin/executable/sequence/chain.go | 9 ++--
plugin/executable/stats_api/stats_api.go | 29 ++++++++++--
plugin/executable/stats_api/stats_api_test.go | 22 +++++++++
4 files changed, 83 insertions(+), 22 deletions(-)
--- a/plugin/executable/forward/forward.go
+++ b/plugin/executable/forward/forward.go
@@ -267,9 +267,11 @@ func (f *Forward) exchange(ctx context.C
done := make(chan struct{})
defer close(done)
+ var lastAttempted *upstreamWrapper
r := rand.IntN(len(us))
for i := 0; i < concurrent; i++ {
u := us[(r+i)%len(us)]
+ lastAttempted = u
qc := copyPayload(queryPayload)
go func(uqid uint32, question dns.Question, chosenUpstream *upstreamWrapper) {
defer pool.ReleaseBuf(qc)
@@ -318,31 +320,44 @@ func (f *Forward) exchange(ctx context.C
}
if chosenUpstream != nil {
- addr := chosenUpstream.cfg.Addr
- proto := "UDP"
- if strings.Contains(addr, "://") {
- parts := strings.SplitN(addr, "://", 2)
- proto = strings.ToUpper(parts[0])
- addr = parts[1]
- }
- if proto == "TLS" {
- proto = "DoT"
- } else if proto == "HTTPS" {
- proto = "DoH"
- } else if proto == "QUIC" || proto == "DOQ" {
- proto = "DoQ"
- }
- qCtx.SetUpstreamSelected(chosenUpstream.cfg.Addr, proto, chosenUpstream.cfg.Tag, f.pluginTag)
+ f.setUpstream(qCtx, chosenUpstream)
}
return r, nil
case <-ctx.Done():
+ if qCtx.UpstreamSelected == nil && lastAttempted != nil {
+ f.setUpstream(qCtx, lastAttempted)
+ }
return nil, context.Cause(ctx)
}
}
+ if qCtx.UpstreamSelected == nil && lastAttempted != nil {
+ f.setUpstream(qCtx, lastAttempted)
+ }
return nil, errors.New("all upstream servers failed")
}
+func (f *Forward) setUpstream(qCtx *query_context.Context, u *upstreamWrapper) {
+ if u == nil {
+ return
+ }
+ addr := u.cfg.Addr
+ proto := "UDP"
+ if strings.Contains(addr, "://") {
+ parts := strings.SplitN(addr, "://", 2)
+ proto = strings.ToUpper(parts[0])
+ addr = parts[1]
+ }
+ if proto == "TLS" {
+ proto = "DoT"
+ } else if proto == "HTTPS" {
+ proto = "DoH"
+ } else if proto == "QUIC" || proto == "DOQ" {
+ proto = "DoQ"
+ }
+ qCtx.SetUpstreamSelected(u.cfg.Addr, proto, u.cfg.Tag, f.pluginTag)
+}
+
func quickSetup(bq sequence.BQ, s string) (any, error) {
args := new(Args)
args.Concurrent = maxConcurrentQueries
--- a/plugin/executable/sequence/chain.go
+++ b/plugin/executable/sequence/chain.go
@@ -122,13 +122,14 @@ checkMatchesLoop:
switch {
case n.E != nil:
hasRespBefore := qCtx.R() != nil
- if err := n.E.Exec(ctx, qCtx); err != nil {
- return err
- }
+ err := n.E.Exec(ctx, qCtx)
hasRespAfter := qCtx.R() != nil
- if shouldLogRule(n, hasRespBefore, hasRespAfter) {
+ if shouldLogRule(n, hasRespBefore, hasRespAfter) || err != nil {
qCtx.AddRuleHit(w.sequenceTag, n.RuleMatches, n.RuleExec)
}
+ if err != nil {
+ return err
+ }
p++
continue
case n.RE != nil:
--- a/plugin/executable/stats_api/stats_api.go
+++ b/plugin/executable/stats_api/stats_api.go
@@ -982,10 +982,33 @@ func (s *StatsAPI) Exec(ctx context.Cont
var answers []AnswerDTO
isBlocked := false
+ isTimeout := false
+ if err != nil {
+ errStr := strings.ToLower(err.Error())
+ if errors.Is(err, context.DeadlineExceeded) ||
+ strings.Contains(errStr, "timeout") ||
+ strings.Contains(errStr, "deadline") ||
+ strings.Contains(errStr, "all upstream servers failed") ||
+ strings.Contains(errStr, "all primary and secondary executables failed") {
+ isTimeout = true
+ }
+ }
+ if elapsed >= 4*time.Second && r == nil {
+ isTimeout = true
+ }
+
if r == nil {
- status = "DROPPED"
- if !hasUpstream {
- isBlocked = true
+ if isTimeout {
+ status = "TIMEOUT"
+ isBlocked = false
+ } else if err != nil {
+ status = "ERROR"
+ isBlocked = false
+ } else {
+ status = "DROPPED"
+ if !hasUpstream {
+ isBlocked = true
+ }
}
} else {
if rcodeStr, ok := dns.RcodeToString[r.Rcode]; ok {
--- a/plugin/executable/stats_api/stats_api_test.go
+++ b/plugin/executable/stats_api/stats_api_test.go
@@ -23,6 +23,7 @@ import (
"bytes"
"context"
"encoding/json"
+ "errors"
"fmt"
"net/http"
"net/http/httptest"
@@ -834,4 +835,25 @@ func TestStatsAPINoAnswerUpstreamAndBloc
if len(logs5) != 1 || !logs5[0].IsBlocked || logs5[0].Status != "DROPPED" {
t.Errorf("expected blocked DROPPED log, got %+v", logs5)
}
+
+ // Case 6: Upstream timeout -> NOT blocked, status is TIMEOUT
+ q6 := new(dns.Msg)
+ q6.SetQuestion("timedout.example.com.", dns.TypeAAAA)
+ qCtx6 := query_context.NewContext(q6)
+ qCtx6.SetUpstreamSelected("114.114.114.114:53", "UDP", "local", "forward_local")
+
+ execTimeout := sequence.ExecutableFunc(func(ctx context.Context, qCtx *query_context.Context) error {
+ return errors.New("all upstream servers failed")
+ })
+
+ walker6 := sequence.NewChainWalker([]*sequence.ChainNode{{E: execTimeout}}, nil)
+ _ = s.Exec(context.Background(), qCtx6, walker6)
+
+ if s.blockedQueries.Load() != 2 {
+ t.Errorf("expected blockedQueries to remain 2, got %d", s.blockedQueries.Load())
+ }
+ _, logs6 := s.ringBuffer.QueryLogs(1, 0, "", "all")
+ if len(logs6) != 1 || logs6[0].IsBlocked || logs6[0].Status != "TIMEOUT" {
+ t.Errorf("expected unblocked TIMEOUT log, got %+v", logs6)
+ }
}
@@ -0,0 +1,630 @@
From 3ce13333944dc6b271b8db3c73f1627c276f839a Mon Sep 17 00:00:00 2001
From: sbwml <admin@cooluc.com>
Date: Thu, 15 Oct 2026 17:58:04 +0800
Subject: [PATCH 3/5] fix: resolve concurrency races, configuration fallbacks,
and parser bugs
- pkg/concurrent_map: fix flush() using RLock instead of Lock causing concurrent map writes
- plugin/executable/black_hole: eliminate global mutex and shuffle local slices to prevent data race
- plugin/executable/sequence/fallback: return early in alwaysStandby to avoid overwriting primary results
- plugin/server: invoke args.init() across server plugins to ensure default listen ports and timeouts
- plugin/data_provider/adblock_set: support multi-domain hosts lines and trim trailing '^' separators
- mlog: restore JSON encoder support when production log mode is enabled
- add comprehensive unit tests and race checks for fallback, black_hole, servers, and mlog
---
mlog/logger.go | 18 ++-
mlog/rotator_test.go | 31 ++++
pkg/concurrent_map/map.go | 4 +-
pkg/concurrent_map/map_test.go | 36 +++++
.../adblock_set/adblock_set_test.go | 7 +
plugin/data_provider/adblock_set/parser.go | 14 +-
plugin/executable/black_hole/black_hole.go | 34 ++--
.../executable/black_hole/black_hole_test.go | 38 +++++
.../executable/sequence/fallback/fallback.go | 4 +-
.../sequence/fallback/fallback_test.go | 146 ++++++++++++++++++
plugin/server/http_server/http_server.go | 3 +-
plugin/server/http_server/http_server_test.go | 13 ++
plugin/server/quic_server/quic_server.go | 1 +
plugin/server/quic_server/quic_server_test.go | 13 ++
plugin/server/tcp_server/tcp_server.go | 1 +
plugin/server/tcp_server/tcp_server_test.go | 14 ++
plugin/server/udp_server/udp_server.go | 1 +
plugin/server/udp_server/udp_server_test.go | 13 ++
18 files changed, 362 insertions(+), 29 deletions(-)
create mode 100644 plugin/executable/black_hole/black_hole_test.go
create mode 100644 plugin/executable/sequence/fallback/fallback_test.go
create mode 100644 plugin/server/http_server/http_server_test.go
create mode 100644 plugin/server/quic_server/quic_server_test.go
create mode 100644 plugin/server/tcp_server/tcp_server_test.go
create mode 100644 plugin/server/udp_server/udp_server_test.go
--- a/mlog/logger.go
+++ b/mlog/logger.go
@@ -93,13 +93,23 @@ func NewLogger(lc LogConfig) (*zap.Logge
out = stderr
}
- encoderConfig := zap.NewDevelopmentEncoderConfig()
- encoderConfig.EncodeTime = func(t time.Time, enc zapcore.PrimitiveArrayEncoder) {
- enc.AppendString(t.Format("2006-01-02 15:04:05"))
+ var encoder zapcore.Encoder
+ if lc.Production {
+ encoderConfig := zap.NewProductionEncoderConfig()
+ encoderConfig.EncodeTime = func(t time.Time, enc zapcore.PrimitiveArrayEncoder) {
+ enc.AppendString(t.Format("2006-01-02 15:04:05"))
+ }
+ encoder = zapcore.NewJSONEncoder(encoderConfig)
+ } else {
+ encoderConfig := zap.NewDevelopmentEncoderConfig()
+ encoderConfig.EncodeTime = func(t time.Time, enc zapcore.PrimitiveArrayEncoder) {
+ enc.AppendString(t.Format("2006-01-02 15:04:05"))
+ }
+ encoder = zapcore.NewConsoleEncoder(encoderConfig)
}
core := zapcore.NewCore(
- zapcore.NewConsoleEncoder(encoderConfig),
+ encoder,
out,
lvl,
)
--- a/mlog/rotator_test.go
+++ b/mlog/rotator_test.go
@@ -1,7 +1,12 @@
package mlog
import (
+ "encoding/json"
+ "os"
+ "path/filepath"
"testing"
+
+ "go.uber.org/zap"
)
func TestParseSize(t *testing.T) {
@@ -33,3 +38,29 @@ func TestNewLoggerBackups(t *testing.T)
t.Fatalf("NewLogger failed: %v", err)
}
}
+
+func TestNewLoggerProduction(t *testing.T) {
+ tmpFile := filepath.Join(t.TempDir(), "test_prod.log")
+ lc := LogConfig{
+ File: tmpFile,
+ Production: true,
+ }
+ logger, err := NewLogger(lc)
+ if err != nil {
+ t.Fatalf("NewLogger failed: %v", err)
+ }
+ logger.Info("hello production", zap.String("key", "val"))
+ _ = logger.Sync()
+
+ data, err := os.ReadFile(tmpFile)
+ if err != nil {
+ t.Fatalf("read log file failed: %v", err)
+ }
+ var m map[string]any
+ if err := json.Unmarshal(data, &m); err != nil {
+ t.Fatalf("expected valid json from production logger, got %s: %v", string(data), err)
+ }
+ if m["key"] != "val" || m["msg"] != "hello production" {
+ t.Fatalf("unexpected json log content: %v", m)
+ }
+}
--- a/pkg/concurrent_map/map.go
+++ b/pkg/concurrent_map/map.go
@@ -162,8 +162,8 @@ func (m *shard[K, V]) len() int {
}
func (m *shard[K, V]) flush() {
- m.l.RLock()
- defer m.l.RUnlock()
+ m.l.Lock()
+ defer m.l.Unlock()
m.m = make(map[K]V)
}
--- a/pkg/concurrent_map/map_test.go
+++ b/pkg/concurrent_map/map_test.go
@@ -139,6 +139,42 @@ func TestConcurrentMap_TestAndSet(t *tes
}
}
+func Test_Map_Flush_Concurrency(t *testing.T) {
+ m := NewMap[testMapHashable, int]()
+ var wg sync.WaitGroup
+ const goroutines = 20
+ const iterations = 500
+
+ wg.Add(goroutines * 2 + 1)
+ // readers
+ for i := 0; i < goroutines; i++ {
+ go func(base int) {
+ defer wg.Done()
+ for j := 0; j < iterations; j++ {
+ _, _ = m.Get(testMapHashable(base + (j % 50)))
+ }
+ }(i * 100)
+ }
+ // writers
+ for i := 0; i < goroutines; i++ {
+ go func(base int) {
+ defer wg.Done()
+ for j := 0; j < iterations; j++ {
+ m.Set(testMapHashable(base+(j%50)), j)
+ }
+ }(i * 100)
+ }
+ // flushers
+ go func() {
+ defer wg.Done()
+ for j := 0; j < iterations/5; j++ {
+ m.Flush()
+ }
+ }()
+
+ wg.Wait()
+}
+
func BenchmarkConcurrentMap_Get_And_Set(b *testing.B) {
keys := make([]testMapHashable, 2048)
m := NewMap[testMapHashable, int]()
--- a/plugin/data_provider/adblock_set/adblock_set_test.go
+++ b/plugin/data_provider/adblock_set/adblock_set_test.go
@@ -45,6 +45,8 @@ domain:ntp.org
full:metrics.icloud.com
0.0.0.0 singlehotmilf.online
127.0.0.1 deezer.deliveryengine.adswizz.com
+0.0.0.0 multihost1.com multihost2.com # comment here
+trailingsymbol.com^
`
blacklist := domain.NewMixMatcher[struct{}]()
whitelist := domain.NewMixMatcher[struct{}]()
@@ -85,6 +87,11 @@ full:metrics.icloud.com
{"sub.singlehotmilf.online", false}, // hosts: exact match only
{"deezer.deliveryengine.adswizz.com", true}, // hosts: 127.0.0.1
{"sub.deezer.deliveryengine.adswizz.com", false}, // hosts: exact match only
+ {"multihost1.com", true},
+ {"multihost2.com", true},
+ {"comment", false},
+ {"trailingsymbol.com", true},
+ {"sub.trailingsymbol.com", true},
}
for _, tt := range tests {
--- a/plugin/data_provider/adblock_set/parser.go
+++ b/plugin/data_provider/adblock_set/parser.go
@@ -37,12 +37,16 @@ func ParseRules(r io.Reader, blacklist,
continue
}
- // Handle hosts file format: "0.0.0.0 domain" or "127.0.0.1 domain"
+ // Handle hosts file format: "0.0.0.0 domain1 domain2" or "127.0.0.1 domain"
if fields := strings.Fields(line); len(fields) >= 2 {
if ip := net.ParseIP(fields[0]); ip != nil {
- host := fields[1]
- if host != "" && host != "localhost" {
- _ = blacklist.Add("full:"+host, struct{}{})
+ for _, host := range fields[1:] {
+ if strings.HasPrefix(host, "#") {
+ break
+ }
+ if host != "" && host != "localhost" {
+ _ = blacklist.Add("full:"+host, struct{}{})
+ }
}
continue
}
@@ -87,6 +91,8 @@ func ParseRules(r io.Reader, blacklist,
}
func parseAndAddRule(rule string, m *domain.MixMatcher[struct{}]) error {
+ rule = strings.TrimSuffix(rule, "^")
+
// 0. Native MosDNS rules: type:pattern
if strings.Contains(rule, ":") {
typ, _, _ := strings.Cut(rule, ":")
--- a/plugin/executable/black_hole/black_hole.go
+++ b/plugin/executable/black_hole/black_hole.go
@@ -25,10 +25,9 @@ import (
"github.com/IrineSistiana/mosdns/v5/pkg/query_context"
"github.com/IrineSistiana/mosdns/v5/plugin/executable/sequence"
"github.com/miekg/dns"
+ "math/rand"
"net/netip"
"strings"
- "math/rand"
- "sync"
)
const PluginType = "black_hole"
@@ -42,7 +41,6 @@ var _ sequence.Executable = (*BlackHole)
type BlackHole struct {
ipv4 []netip.Addr
ipv6 []netip.Addr
- shuffleMutex sync.Mutex
}
// QuickSetup format: [ipv4|ipv6] ...
@@ -68,21 +66,9 @@ func NewBlackHole(ips []string) (*BlackH
return b, nil
}
-func (b *BlackHole) shuffleIPs() {
- b.shuffleMutex.Lock()
- defer b.shuffleMutex.Unlock()
- rand.Shuffle(len(b.ipv4), func(i, j int) {
- b.ipv4[i], b.ipv4[j] = b.ipv4[j], b.ipv4[i]
- })
- rand.Shuffle(len(b.ipv6), func(i, j int) {
- b.ipv6[i], b.ipv6[j] = b.ipv6[j], b.ipv6[i]
- })
-}
-
// Exec implements sequence.Executable. It set a response with given ips if
// query has corresponding qtypes.
func (b *BlackHole) Exec(_ context.Context, qCtx *query_context.Context) error {
- b.shuffleIPs()
if r := b.Response(qCtx.Q()); r != nil {
qCtx.SetResponse(r)
}
@@ -103,7 +89,14 @@ func (b *BlackHole) Response(q *dns.Msg)
case qtype == dns.TypeA && len(b.ipv4) > 0:
r := new(dns.Msg)
r.SetReply(q)
- for _, addr := range b.ipv4 {
+ ips := b.ipv4
+ if len(ips) > 1 {
+ ips = append([]netip.Addr(nil), b.ipv4...)
+ rand.Shuffle(len(ips), func(i, j int) {
+ ips[i], ips[j] = ips[j], ips[i]
+ })
+ }
+ for _, addr := range ips {
rr := &dns.A{
Hdr: dns.RR_Header{
Name: qName,
@@ -120,7 +113,14 @@ func (b *BlackHole) Response(q *dns.Msg)
case qtype == dns.TypeAAAA && len(b.ipv6) > 0:
r := new(dns.Msg)
r.SetReply(q)
- for _, addr := range b.ipv6 {
+ ips := b.ipv6
+ if len(ips) > 1 {
+ ips = append([]netip.Addr(nil), b.ipv6...)
+ rand.Shuffle(len(ips), func(i, j int) {
+ ips[i], ips[j] = ips[j], ips[i]
+ })
+ }
+ for _, addr := range ips {
rr := &dns.AAAA{
Hdr: dns.RR_Header{
Name: qName,
--- /dev/null
+++ b/plugin/executable/black_hole/black_hole_test.go
@@ -0,0 +1,38 @@
+package black_hole
+
+import (
+ "context"
+ "sync"
+ "testing"
+
+ "github.com/IrineSistiana/mosdns/v5/pkg/query_context"
+ "github.com/miekg/dns"
+ "github.com/stretchr/testify/assert"
+)
+
+func TestBlackHoleConcurrency(t *testing.T) {
+ bh, err := NewBlackHole([]string{"10.0.0.1", "10.0.0.2", "10.0.0.3", "10.0.0.4"})
+ assert.NoError(t, err)
+
+ const goroutines = 50
+ const queriesPerGoroutine = 100
+ var wg sync.WaitGroup
+ wg.Add(goroutines)
+
+ for i := 0; i < goroutines; i++ {
+ go func() {
+ defer wg.Done()
+ for j := 0; j < queriesPerGoroutine; j++ {
+ q := new(dns.Msg)
+ q.SetQuestion("test.example.com.", dns.TypeA)
+ qCtx := query_context.NewContext(q)
+ err := bh.Exec(context.Background(), qCtx)
+ assert.NoError(t, err)
+ r := qCtx.R()
+ assert.NotNil(t, r)
+ assert.Equal(t, 4, len(r.Answer))
+ }
+ }()
+ }
+ wg.Wait()
+}
--- a/plugin/executable/sequence/fallback/fallback.go
+++ b/plugin/executable/sequence/fallback/fallback.go
@@ -133,8 +133,8 @@ func (f *fallback) doFallback(ctx contex
close(primFailed)
respChan <- nil
} else {
- close(primDone)
respChan <- &fallbackResult{r: r, qCtx: qCtx}
+ close(primDone)
}
}()
@@ -167,7 +167,9 @@ func (f *fallback) doFallback(ctx contex
if f.alwaysStandby && r != nil {
select {
case <-ctx.Done():
+ return
case <-primDone:
+ return
case <-primFailed: // only send secondary result when primary is failed.
case <-timer.C: // or timed out.
}
--- /dev/null
+++ b/plugin/executable/sequence/fallback/fallback_test.go
@@ -0,0 +1,146 @@
+package fallback
+
+import (
+ "context"
+ "errors"
+ "testing"
+ "time"
+
+ "github.com/IrineSistiana/mosdns/v5/pkg/query_context"
+ "github.com/miekg/dns"
+ "github.com/stretchr/testify/assert"
+ "go.uber.org/zap"
+)
+
+type mockExec struct {
+ delay time.Duration
+ resp *dns.Msg
+ err error
+}
+
+func (m *mockExec) Exec(ctx context.Context, qCtx *query_context.Context) error {
+ if m.delay > 0 {
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-time.After(m.delay):
+ }
+ }
+ if m.err != nil {
+ return m.err
+ }
+ if m.resp != nil {
+ qCtx.SetResponse(m.resp.Copy())
+ }
+ return nil
+}
+
+func TestFallbackPrimarySuccess(t *testing.T) {
+ respP := new(dns.Msg)
+ respP.SetRcode(new(dns.Msg), dns.RcodeSuccess)
+ respP.Extra = append(respP.Extra, &dns.TXT{Hdr: dns.RR_Header{Name: "prim.", Rrtype: dns.TypeTXT, Class: dns.ClassINET, Ttl: 60}, Txt: []string{"primary"}})
+
+ respS := new(dns.Msg)
+ respS.SetRcode(new(dns.Msg), dns.RcodeSuccess)
+ respS.Extra = append(respS.Extra, &dns.TXT{Hdr: dns.RR_Header{Name: "sec.", Rrtype: dns.TypeTXT, Class: dns.ClassINET, Ttl: 60}, Txt: []string{"secondary"}})
+
+ fb := &fallback{
+ logger: zap.NewNop(),
+ primary: &mockExec{resp: respP},
+ secondary: &mockExec{resp: respS, delay: 50 * time.Millisecond},
+ fastFallbackDuration: 100 * time.Millisecond,
+ }
+
+ q := new(dns.Msg)
+ q.SetQuestion("example.com.", dns.TypeA)
+ qCtx := query_context.NewContext(q)
+
+ err := fb.Exec(context.Background(), qCtx)
+ assert.NoError(t, err)
+ assert.NotNil(t, qCtx.R())
+ assert.Equal(t, "prim.", qCtx.R().Extra[0].Header().Name)
+}
+
+func TestFallbackSecondaryOnPrimaryFail(t *testing.T) {
+ respS := new(dns.Msg)
+ respS.SetRcode(new(dns.Msg), dns.RcodeSuccess)
+
+ fb := &fallback{
+ logger: zap.NewNop(),
+ primary: &mockExec{err: errors.New("upstream timeout")},
+ secondary: &mockExec{resp: respS},
+ fastFallbackDuration: 100 * time.Millisecond,
+ }
+
+ q := new(dns.Msg)
+ q.SetQuestion("example.com.", dns.TypeA)
+ qCtx := query_context.NewContext(q)
+
+ err := fb.Exec(context.Background(), qCtx)
+ assert.NoError(t, err)
+ assert.NotNil(t, qCtx.R())
+}
+
+func TestFallbackSecondaryOnPrimaryTimeout(t *testing.T) {
+ respS := new(dns.Msg)
+ respS.SetRcode(new(dns.Msg), dns.RcodeSuccess)
+
+ fb := &fallback{
+ logger: zap.NewNop(),
+ primary: &mockExec{delay: 200 * time.Millisecond, err: errors.New("timeout")},
+ secondary: &mockExec{resp: respS},
+ fastFallbackDuration: 20 * time.Millisecond,
+ }
+
+ q := new(dns.Msg)
+ q.SetQuestion("example.com.", dns.TypeA)
+ qCtx := query_context.NewContext(q)
+
+ err := fb.Exec(context.Background(), qCtx)
+ assert.NoError(t, err)
+ assert.NotNil(t, qCtx.R())
+}
+
+func TestFallbackBothFail(t *testing.T) {
+ fb := &fallback{
+ logger: zap.NewNop(),
+ primary: &mockExec{err: errors.New("primary failed")},
+ secondary: &mockExec{err: errors.New("secondary failed")},
+ fastFallbackDuration: 20 * time.Millisecond,
+ }
+
+ q := new(dns.Msg)
+ q.SetQuestion("example.com.", dns.TypeA)
+ qCtx := query_context.NewContext(q)
+
+ err := fb.Exec(context.Background(), qCtx)
+ assert.ErrorIs(t, err, ErrFailed)
+}
+
+func TestFallbackAlwaysStandbyPrimaryWins(t *testing.T) {
+ respP := new(dns.Msg)
+ respP.SetRcode(new(dns.Msg), dns.RcodeSuccess)
+ respP.Extra = append(respP.Extra, &dns.TXT{Hdr: dns.RR_Header{Name: "prim.", Rrtype: dns.TypeTXT, Class: dns.ClassINET, Ttl: 60}, Txt: []string{"primary"}})
+
+ respS := new(dns.Msg)
+ respS.SetRcode(new(dns.Msg), dns.RcodeSuccess)
+ respS.Extra = append(respS.Extra, &dns.TXT{Hdr: dns.RR_Header{Name: "sec.", Rrtype: dns.TypeTXT, Class: dns.ClassINET, Ttl: 60}, Txt: []string{"secondary"}})
+
+ // Secondary finishes faster (5ms), but primary succeeds before threshold (50ms)
+ fb := &fallback{
+ logger: zap.NewNop(),
+ primary: &mockExec{delay: 15 * time.Millisecond, resp: respP},
+ secondary: &mockExec{delay: 2 * time.Millisecond, resp: respS},
+ fastFallbackDuration: 50 * time.Millisecond,
+ alwaysStandby: true,
+ }
+
+ q := new(dns.Msg)
+ q.SetQuestion("example.com.", dns.TypeA)
+ qCtx := query_context.NewContext(q)
+
+ err := fb.Exec(context.Background(), qCtx)
+ assert.NoError(t, err)
+ assert.NotNil(t, qCtx.R())
+ assert.Equal(t, "prim.", qCtx.R().Extra[0].Header().Name)
+}
--- a/plugin/server/http_server/http_server.go
+++ b/plugin/server/http_server/http_server.go
@@ -17,7 +17,7 @@
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
-package tcp_server
+package http_server
import (
"context"
@@ -72,6 +72,7 @@ func Init(bp *coremain.BP, args any) (an
}
func StartServer(bp *coremain.BP, args *Args) (*HttpServer, error) {
+ args.init()
mux := http.NewServeMux()
for _, entry := range args.Entries {
dh, err := server_utils.NewHandler(bp, entry.Exec)
--- /dev/null
+++ b/plugin/server/http_server/http_server_test.go
@@ -0,0 +1,13 @@
+package http_server
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func TestArgsInit(t *testing.T) {
+ a := new(Args)
+ a.init()
+ assert.Equal(t, 30, a.IdleTimeout)
+}
--- a/plugin/server/quic_server/quic_server.go
+++ b/plugin/server/quic_server/quic_server.go
@@ -67,6 +67,7 @@ func Init(bp *coremain.BP, args any) (an
}
func StartServer(bp *coremain.BP, args *Args) (*QuicServer, error) {
+ args.init()
logger := bp.L()
dh, err := server_utils.NewHandler(bp, args.Entry)
--- /dev/null
+++ b/plugin/server/quic_server/quic_server_test.go
@@ -0,0 +1,13 @@
+package quic_server
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func TestArgsInit(t *testing.T) {
+ a := new(Args)
+ a.init()
+ assert.Equal(t, 30, a.IdleTimeout)
+}
--- a/plugin/server/tcp_server/tcp_server.go
+++ b/plugin/server/tcp_server/tcp_server.go
@@ -68,6 +68,7 @@ func Init(bp *coremain.BP, args any) (an
}
func StartServer(bp *coremain.BP, args *Args) (*TcpServer, error) {
+ args.init()
dh, err := server_utils.NewHandler(bp, args.Entry)
if err != nil {
return nil, fmt.Errorf("failed to init dns handler, %w", err)
--- /dev/null
+++ b/plugin/server/tcp_server/tcp_server_test.go
@@ -0,0 +1,14 @@
+package tcp_server
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func TestArgsInit(t *testing.T) {
+ a := new(Args)
+ a.init()
+ assert.Equal(t, "127.0.0.1:53", a.Listen)
+ assert.Equal(t, 10, a.IdleTimeout)
+}
--- a/plugin/server/udp_server/udp_server.go
+++ b/plugin/server/udp_server/udp_server.go
@@ -61,6 +61,7 @@ func Init(bp *coremain.BP, args any) (an
}
func StartServer(bp *coremain.BP, args *Args) (*UdpServer, error) {
+ args.init()
dh, err := server_utils.NewHandler(bp, args.Entry)
if err != nil {
return nil, fmt.Errorf("failed to init dns handler, %w", err)
--- /dev/null
+++ b/plugin/server/udp_server/udp_server_test.go
@@ -0,0 +1,13 @@
+package udp_server
+
+import (
+ "testing"
+
+ "github.com/stretchr/testify/assert"
+)
+
+func TestArgsInit(t *testing.T) {
+ a := new(Args)
+ a.init()
+ assert.Equal(t, "127.0.0.1:53", a.Listen)
+}
@@ -0,0 +1,786 @@
From 9425cf5e6520b2381175fe4d69f15fcbf4dbe256 Mon Sep 17 00:00:00 2001
From: sbwml <admin@cooluc.com>
Date: Thu, 15 Oct 2026 18:59:58 +0800
Subject: [PATCH 4/5] fix(stats_api): fix loopback false-positive blocking,
null answers and optimize memory
- Exclude loopback domain queries (localhost) from ad-blocking while preserving adblock sinkhole rules
- Fix empty answers slice serializing as null instead of empty JSON array
- Exclude RFC 9460 SVCB queries from false-positive ad-blocking on NXDOMAIN
- Connect handleClearCache to FlushAllCaches on cache and dual_selector plugins
- Persist state atomically with fsync and rename to prevent corrupted dumps
- Immediately flush disk dump upon clearing logs
- Optimize RingBuffer and TopStats memory allocations and enforce 20k entry bounds
- Align history retention window to 168 hours (7 days)
---
coremain/mosdns.go | 13 +
plugin/executable/cache/cache.go | 5 +
.../executable/dual_selector/dual_selector.go | 5 +
plugin/executable/stats_api/stats_api.go | 265 +++++++++++++-----
plugin/executable/stats_api/stats_api_test.go | 206 ++++++++++++++
5 files changed, 423 insertions(+), 71 deletions(-)
--- a/coremain/mosdns.go
+++ b/coremain/mosdns.go
@@ -152,6 +152,19 @@ func (m *Mosdns) GetPlugin(tag string) a
return m.plugins[tag]
}
+type Flusher interface {
+ Flush()
+}
+
+// FlushAllCaches flushes all registered plugins that implement Flusher.
+func (m *Mosdns) FlushAllCaches() {
+ for _, p := range m.plugins {
+ if f, ok := p.(Flusher); ok {
+ f.Flush()
+ }
+ }
+}
+
// GetMetricsReg returns a prometheus.Registerer with a prefix of "mosdns_"
func (m *Mosdns) GetMetricsReg() prometheus.Registerer {
return prometheus.WrapRegistererWithPrefix("mosdns_", m.metricsReg)
--- a/plugin/executable/cache/cache.go
+++ b/plugin/executable/cache/cache.go
@@ -373,6 +373,11 @@ func (c *Cache) Close() error {
return c.backend.Close()
}
+// Flush removes all stored entries from backend.
+func (c *Cache) Flush() {
+ c.backend.Flush()
+}
+
func (c *Cache) loadDump() error {
if len(c.args.DumpFile) == 0 {
return nil
--- a/plugin/executable/dual_selector/dual_selector.go
+++ b/plugin/executable/dual_selector/dual_selector.go
@@ -172,6 +172,11 @@ func (s *Selector) Close() error {
return nil
}
+// Flush removes all cached entries from preferTypOkCache.
+func (s *Selector) Flush() {
+ s.preferTypOkCache.Flush()
+}
+
func NewPreferIpv4(bq sequence.BQ) *Selector {
return newSelector(bq, dns.TypeA)
}
--- a/plugin/executable/stats_api/stats_api.go
+++ b/plugin/executable/stats_api/stats_api.go
@@ -28,6 +28,7 @@ import (
"math"
"net/http"
"os"
+ "path/filepath"
"sort"
"strconv"
"strings"
@@ -137,25 +138,48 @@ func (r *RingBuffer) Clear() {
}
func (r *RingBuffer) QueryLogs(limit, offset int, search, filter string) (int, []LogEntry) {
+ searchLower := strings.ToLower(strings.TrimSpace(search))
+ filterLower := strings.ToLower(strings.TrimSpace(filter))
+
r.mu.RLock()
defer r.mu.RUnlock()
- all := make([]LogEntry, 0, r.count)
+ filtered := make([]LogEntry, 0, r.count)
for i := 0; i < r.count; i++ {
idx := (r.head - 1 - i + r.capacity) % r.capacity
- all = append(all, r.buf[idx])
- }
+ entry := r.buf[idx]
- searchLower := strings.ToLower(strings.TrimSpace(search))
- filterLower := strings.ToLower(strings.TrimSpace(filter))
-
- filtered := make([]LogEntry, 0, len(all))
- for _, entry := range all {
- if filterLower == "blocked" && !entry.IsBlocked {
- continue
- }
- if filterLower == "cached" && !entry.IsCached {
- continue
+ switch filterLower {
+ case "blocked":
+ if !entry.IsBlocked {
+ continue
+ }
+ case "cached":
+ if !entry.IsCached {
+ continue
+ }
+ case "hosts":
+ if entry.Upstream != "hosts" && entry.Rule != "hosts" {
+ continue
+ }
+ case "arbitrary":
+ if entry.Upstream != "arbitrary" && entry.Rule != "arbitrary" {
+ continue
+ }
+ case "timeout":
+ if entry.Status != "TIMEOUT" {
+ continue
+ }
+ case "error":
+ if entry.Status != "ERROR" {
+ continue
+ }
+ case "all", "":
+ // no filter
+ default:
+ if !strings.EqualFold(entry.Status, filterLower) {
+ continue
+ }
}
if searchLower != "" {
@@ -241,19 +265,49 @@ func NewTopStats() *TopStats {
}
}
+const maxTopEntries = 20000
+
+func pruneMap(m map[string]uint64) {
+ if len(m) <= maxTopEntries {
+ return
+ }
+ for k, v := range m {
+ if v <= 1 {
+ delete(m, k)
+ }
+ }
+ if len(m) > maxTopEntries {
+ for k, v := range m {
+ if v <= 2 {
+ delete(m, k)
+ }
+ }
+ }
+}
+
func (t *TopStats) Record(domain, clientIP string, isBlocked bool) {
t.mu.Lock()
defer t.mu.Unlock()
if domain != "" {
+ domainLower := strings.ToLower(domain)
if isBlocked {
- t.topBlocked[domain]++
+ t.topBlocked[domainLower]++
+ if len(t.topBlocked) > maxTopEntries {
+ pruneMap(t.topBlocked)
+ }
} else {
- t.topDomains[domain]++
+ t.topDomains[domainLower]++
+ if len(t.topDomains) > maxTopEntries {
+ pruneMap(t.topDomains)
+ }
}
}
if clientIP != "" {
t.topClients[clientIP]++
+ if len(t.topClients) > maxTopEntries {
+ pruneMap(t.topClients)
+ }
}
}
@@ -303,38 +357,6 @@ func (t *TopStats) Import(domains, clien
}
}
-func getSortedTop(m map[string]uint64, isClient bool, limit int) []TopItem {
- type pair struct {
- key string
- count uint64
- }
- pairs := make([]pair, 0, len(m))
- for k, v := range m {
- pairs = append(pairs, pair{key: k, count: v})
- }
- sort.Slice(pairs, func(i, j int) bool {
- if pairs[i].count == pairs[j].count {
- return pairs[i].key < pairs[j].key
- }
- return pairs[i].count > pairs[j].count
- })
-
- if limit > len(pairs) {
- limit = len(pairs)
- }
- res := make([]TopItem, 0, limit)
- for i := 0; i < limit; i++ {
- item := TopItem{Count: pairs[i].count}
- if isClient {
- item.ClientIP = pairs[i].key
- } else {
- item.Domain = pairs[i].key
- }
- res = append(res, item)
- }
- return res
-}
-
func (t *TopStats) GetTop(limit int) ([]TopItem, []TopItem, []TopItem) {
t.mu.RLock()
defer t.mu.RUnlock()
@@ -343,16 +365,56 @@ func (t *TopStats) GetTop(limit int) ([]
limit = 10
}
- cleanTopDomains := make(map[string]uint64, len(t.topDomains))
+ type pair struct {
+ key string
+ count uint64
+ }
+
+ sortPairs := func(pairs []pair, isClient bool) []TopItem {
+ sort.Slice(pairs, func(i, j int) bool {
+ if pairs[i].count == pairs[j].count {
+ return pairs[i].key < pairs[j].key
+ }
+ return pairs[i].count > pairs[j].count
+ })
+
+ n := limit
+ if n > len(pairs) {
+ n = len(pairs)
+ }
+ res := make([]TopItem, 0, n)
+ for i := 0; i < n; i++ {
+ item := TopItem{Count: pairs[i].count}
+ if isClient {
+ item.ClientIP = pairs[i].key
+ } else {
+ item.Domain = pairs[i].key
+ }
+ res = append(res, item)
+ }
+ return res
+ }
+
+ domainPairs := make([]pair, 0, len(t.topDomains))
for domain, count := range t.topDomains {
if _, isBlocked := t.topBlocked[domain]; !isBlocked {
- cleanTopDomains[domain] = count
+ domainPairs = append(domainPairs, pair{key: domain, count: count})
}
}
- topDomains := getSortedTop(cleanTopDomains, false, limit)
- topClients := getSortedTop(t.topClients, true, limit)
- topBlocked := getSortedTop(t.topBlocked, false, limit)
+ clientPairs := make([]pair, 0, len(t.topClients))
+ for client, count := range t.topClients {
+ clientPairs = append(clientPairs, pair{key: client, count: count})
+ }
+
+ blockedPairs := make([]pair, 0, len(t.topBlocked))
+ for blocked, count := range t.topBlocked {
+ blockedPairs = append(blockedPairs, pair{key: blocked, count: count})
+ }
+
+ topDomains := sortPairs(domainPairs, false)
+ topClients := sortPairs(clientPairs, true)
+ topBlocked := sortPairs(blockedPairs, false)
return topDomains, topClients, topBlocked
}
@@ -400,8 +462,8 @@ func (h *HistoryStats) Record(t time.Tim
bucket = &HistoryBucket{}
h.points[tHour] = bucket
- // Clean up old buckets beyond 48 hours
- cutoff := t.Add(-48 * time.Hour).Unix()
+ // Clean up old buckets beyond 168 hours (7 days)
+ cutoff := t.Add(-168 * time.Hour).Unix()
for k := range h.points {
if k < cutoff {
delete(h.points, k)
@@ -474,7 +536,7 @@ func (h *HistoryStats) Import(points map
defer h.mu.Unlock()
h.points = make(map[int64]*HistoryBucket, len(points))
- cutoff := time.Now().Add(-48 * time.Hour).Unix()
+ cutoff := time.Now().Add(-168 * time.Hour).Unix()
for k, v := range points {
if k >= cutoff {
bucket := &HistoryBucket{}
@@ -506,6 +568,8 @@ type StatsDumpData struct {
type StatsAPI struct {
args *Args
logger *zap.Logger
+ m *coremain.Mosdns
+ tag string
ringBuffer *RingBuffer
topStats *TopStats
@@ -528,6 +592,8 @@ type StatsAPI struct {
func Init(bp *coremain.BP, args any) (any, error) {
a := args.(*Args)
s := NewStatsAPI(a, bp.L())
+ s.m = bp.M()
+ s.tag = bp.Tag()
bp.RegAPI(s.Router())
return s, nil
}
@@ -554,12 +620,14 @@ func QuickSetup(bq sequence.BQ, s string
dumpInterval = d
}
}
- return NewStatsAPI(&Args{
+ api := NewStatsAPI(&Args{
Listen: listen,
Capacity: capacity,
DumpFile: dumpFile,
DumpInterval: dumpInterval,
- }, bq.L()), nil
+ }, bq.L())
+ api.m = bq.M()
+ return api, nil
}
func NewStatsAPI(args *Args, logger *zap.Logger) *StatsAPI {
@@ -584,8 +652,9 @@ func NewStatsAPI(args *Args, logger *zap
if len(args.Listen) > 0 {
srv := &http.Server{
- Addr: args.Listen,
- Handler: s.Router(),
+ Addr: args.Listen,
+ Handler: s.Router(),
+ ReadHeaderTimeout: 5 * time.Second,
}
s.httpServer = srv
go func() {
@@ -623,6 +692,16 @@ func (s *StatsAPI) Router() *chi.Mux {
r.Post("/api/v1/logs/clear", s.handleClearLogs)
r.Post("/api/v1/cache/clear", s.handleClearCache)
+ // Relative endpoints when mounted under a prefix (e.g. /plugins/<tag>/stats)
+ r.Get("/stats", s.handleStats)
+ r.Get("/logs", s.handleLogs)
+ r.Get("/top", s.handleTop)
+ r.Get("/history", s.handleHistory)
+ r.Get("/dump", s.handleDump)
+ r.Post("/load_dump", s.handleLoadDump)
+ r.Post("/logs/clear", s.handleClearLogs)
+ r.Post("/cache/clear", s.handleClearCache)
+
return r
}
@@ -637,6 +716,9 @@ func (s *StatsAPI) handleStats(w http.Re
var blockedPct, cachedPct, avgLat float64
if total > 0 {
blockedPct = float64(blocked) / float64(total) * 100.0
+ if blockedPct > 100.0 {
+ blockedPct = 100.0
+ }
avgLat = (float64(latUs) / float64(total)) / 1000.0
}
excluded := blocked + hosts + arbitrary
@@ -751,7 +833,7 @@ func (s *StatsAPI) handleDump(w http.Res
}
func (s *StatsAPI) handleLoadDump(w http.ResponseWriter, req *http.Request) {
- if err := s.readDump(req.Body); err != nil {
+ if err := s.readDump(io.LimitReader(req.Body, 100<<20)); err != nil {
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
@@ -763,6 +845,7 @@ func (s *StatsAPI) handleClearLogs(w htt
s.ringBuffer.Clear()
s.topStats.Clear()
s.updatedCount.Add(1)
+ _ = s.dumpStats()
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]any{
@@ -772,6 +855,9 @@ func (s *StatsAPI) handleClearLogs(w htt
}
func (s *StatsAPI) handleClearCache(w http.ResponseWriter, req *http.Request) {
+ if s.m != nil {
+ s.m.FlushAllCaches()
+ }
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]any{
"success": true,
@@ -864,15 +950,35 @@ func (s *StatsAPI) dumpStats() error {
if len(s.args.DumpFile) == 0 {
return nil
}
- f, err := os.Create(s.args.DumpFile)
+ if dir := filepath.Dir(s.args.DumpFile); dir != "" && dir != "." {
+ _ = os.MkdirAll(dir, 0755)
+ }
+ tmpFile := fmt.Sprintf("%s.tmp.%d", s.args.DumpFile, time.Now().UnixNano())
+ f, err := os.Create(tmpFile)
if err != nil {
return err
}
- defer f.Close()
if err := s.writeDump(f); err != nil {
+ _ = f.Close()
+ _ = os.Remove(tmpFile)
return fmt.Errorf("failed to write stats dump, %w", err)
}
+ if err := f.Sync(); err != nil {
+ _ = f.Close()
+ _ = os.Remove(tmpFile)
+ return fmt.Errorf("failed to sync stats dump, %w", err)
+ }
+ if err := f.Close(); err != nil {
+ _ = os.Remove(tmpFile)
+ return fmt.Errorf("failed to close stats dump, %w", err)
+ }
+
+ if err := os.Rename(tmpFile, s.args.DumpFile); err != nil {
+ _ = os.Remove(tmpFile)
+ return fmt.Errorf("failed to commit stats dump: %w", err)
+ }
+
s.logger.Info("stats dumped successfully", zap.String("file", s.args.DumpFile))
return nil
}
@@ -890,6 +996,7 @@ func (s *StatsAPI) startDumpLoop() {
if s.updatedCount.Swap(0) > 0 {
if err := s.dumpStats(); err != nil {
s.logger.Error("failed to dump stats", zap.Error(err))
+ s.updatedCount.Add(1)
}
}
case <-s.closeNotify:
@@ -946,6 +1053,15 @@ func (s *StatsAPI) startQPSLoop() {
}()
}
+func isLoopbackDomain(domain string) bool {
+ d := strings.ToLower(strings.TrimSuffix(domain, "."))
+ return d == "localhost" ||
+ strings.HasSuffix(d, ".localhost") ||
+ d == "localhost.localdomain" ||
+ d == "ip6-localhost" ||
+ d == "ip6-loopback"
+}
+
func (s *StatsAPI) Exec(ctx context.Context, qCtx *query_context.Context, next sequence.ChainWalker) error {
start := time.Now()
err := next.ExecNext(ctx, qCtx)
@@ -979,7 +1095,7 @@ func (s *StatsAPI) Exec(ctx context.Cont
r := qCtx.R()
var status string
- var answers []AnswerDTO
+ answers := make([]AnswerDTO, 0)
isBlocked := false
isTimeout := false
@@ -1017,7 +1133,8 @@ func (s *StatsAPI) Exec(ctx context.Cont
status = fmt.Sprintf("RCODE%d", r.Rcode)
}
- if !hasUpstream && (r.Rcode == dns.RcodeNameError || r.Rcode == dns.RcodeRefused) && qQuestion.Qtype != dns.TypeHTTPS {
+ if !hasUpstream && (r.Rcode == dns.RcodeNameError || r.Rcode == dns.RcodeRefused) &&
+ qQuestion.Qtype != dns.TypeHTTPS && qQuestion.Qtype != dns.TypeSVCB {
isBlocked = true
}
@@ -1031,12 +1148,12 @@ func (s *StatsAPI) Exec(ctx context.Cont
switch record := rr.(type) {
case *dns.A:
dataStr = record.A.String()
- if dataStr == "0.0.0.0" || dataStr == "127.0.0.1" {
+ if dataStr == "0.0.0.0" || (dataStr == "127.0.0.1" && !isLoopbackDomain(domain)) {
isBlocked = true
}
case *dns.AAAA:
dataStr = record.AAAA.String()
- if dataStr == "::" || dataStr == "::1" {
+ if dataStr == "::" || (dataStr == "::1" && !isLoopbackDomain(domain)) {
isBlocked = true
}
case *dns.CNAME:
@@ -1047,6 +1164,10 @@ func (s *StatsAPI) Exec(ctx context.Cont
dataStr = record.Ptr
case *dns.MX:
dataStr = record.Mx
+ case *dns.NS:
+ dataStr = record.Ns
+ case *dns.SRV:
+ dataStr = fmt.Sprintf("%d %d %d %s", record.Priority, record.Weight, record.Port, record.Target)
default:
dataStr = rr.String()
}
@@ -1123,12 +1244,12 @@ func (s *StatsAPI) Exec(ctx context.Cont
break
}
- if exec != "" && exec != "$stats_collector" {
+ if exec != "" && exec != "$stats_collector" && exec != "stats_api" && (s.tag == "" || exec != s.tag) {
rule = exec
break
}
- if hit.Sequence != "" && hit.Sequence != "has_resp_sequence" && hit.Sequence != "main_sequence" {
+ if hit.Sequence != "" && hit.Sequence != "has_resp_sequence" && hit.Sequence != "main_sequence" && (s.tag == "" || hit.Sequence != s.tag) {
rule = hit.Sequence
break
}
@@ -1141,6 +1262,8 @@ func (s *StatsAPI) Exec(ctx context.Cont
rule = "hosts"
} else if isArbitrary {
rule = "arbitrary"
+ } else if isBlocked {
+ rule = "block"
} else {
rule = "-"
}
@@ -1173,14 +1296,14 @@ func (s *StatsAPI) Exec(ctx context.Cont
func (s *StatsAPI) Close() error {
s.closeOnce.Do(func() {
close(s.closeNotify)
- if err := s.dumpStats(); err != nil {
- s.logger.Error("failed to dump stats on close", zap.Error(err))
- }
if s.httpServer != nil {
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()
_ = s.httpServer.Shutdown(ctx)
}
+ if err := s.dumpStats(); err != nil {
+ s.logger.Error("failed to dump stats on close", zap.Error(err))
+ }
})
return nil
}
--- a/plugin/executable/stats_api/stats_api_test.go
+++ b/plugin/executable/stats_api/stats_api_test.go
@@ -29,6 +29,7 @@ import (
"net/http/httptest"
"os"
"path/filepath"
+ "strings"
"testing"
"time"
@@ -857,3 +858,208 @@ func TestStatsAPINoAnswerUpstreamAndBloc
t.Errorf("expected unblocked TIMEOUT log, got %+v", logs6)
}
}
+
+func TestStatsAPIAnswersNotNull(t *testing.T) {
+ s := NewStatsAPI(&Args{Capacity: 10}, zap.NewNop())
+ q := new(dns.Msg)
+ q.SetQuestion("nodata.example.com.", dns.TypeA)
+ qCtx := query_context.NewContext(q)
+
+ execEmpty := sequence.ExecutableFunc(func(ctx context.Context, qCtx *query_context.Context) error {
+ resp := new(dns.Msg)
+ resp.SetReply(qCtx.Q())
+ // Empty Answer slice
+ qCtx.SetResponse(resp)
+ return nil
+ })
+
+ walker := sequence.NewChainWalker([]*sequence.ChainNode{{E: execEmpty}}, nil)
+ if err := s.Exec(context.Background(), qCtx, walker); err != nil {
+ t.Fatalf("Exec failed: %v", err)
+ }
+
+ _, logs := s.ringBuffer.QueryLogs(1, 0, "", "all")
+ if len(logs) != 1 {
+ t.Fatalf("expected 1 log entry, got %d", len(logs))
+ }
+ if logs[0].Answers == nil {
+ t.Errorf("expected non-nil answers slice, got nil")
+ }
+
+ rawJSON, err := json.Marshal(logs[0])
+ if err != nil {
+ t.Fatalf("failed to marshal log entry: %v", err)
+ }
+ if !strings.Contains(string(rawJSON), `"answers":[]`) {
+ t.Errorf("expected answers:[] in JSON, got: %s", string(rawJSON))
+ }
+}
+
+func TestStatsAPISVCBNotBlocked(t *testing.T) {
+ s := NewStatsAPI(&Args{Capacity: 10}, zap.NewNop())
+ q := new(dns.Msg)
+ q.SetQuestion("example.com.", dns.TypeSVCB)
+ qCtx := query_context.NewContext(q)
+
+ execNX := sequence.ExecutableFunc(func(ctx context.Context, qCtx *query_context.Context) error {
+ resp := new(dns.Msg)
+ resp.SetReply(qCtx.Q())
+ resp.Rcode = dns.RcodeNameError
+ qCtx.SetResponse(resp)
+ return nil
+ })
+
+ walker := sequence.NewChainWalker([]*sequence.ChainNode{{E: execNX}}, nil)
+ if err := s.Exec(context.Background(), qCtx, walker); err != nil {
+ t.Fatalf("Exec failed: %v", err)
+ }
+
+ if s.blockedQueries.Load() != 0 {
+ t.Errorf("expected blockedQueries 0 for SVCB NXDOMAIN, got %d", s.blockedQueries.Load())
+ }
+}
+
+func TestStatsAPILoopbackResolution(t *testing.T) {
+ s := NewStatsAPI(&Args{Capacity: 10}, zap.NewNop())
+
+ // 1. Hosts returning 127.0.0.1 (e.g. localhost) -> NOT blocked, hostsQueries=1
+ q1 := new(dns.Msg)
+ q1.SetQuestion("localhost.", dns.TypeA)
+ qCtx1 := query_context.NewContext(q1)
+
+ execHosts127 := sequence.ExecutableFunc(func(ctx context.Context, qCtx *query_context.Context) error {
+ resp := new(dns.Msg)
+ resp.SetReply(qCtx.Q())
+ resp.Answer = append(resp.Answer, &dns.A{
+ Hdr: dns.RR_Header{Name: "localhost.", Rrtype: dns.TypeA, Class: dns.ClassINET, Ttl: 300},
+ A: []byte{127, 0, 0, 1},
+ })
+ qCtx.SetResponse(resp)
+ qCtx.SetFromHosts()
+ return nil
+ })
+
+ walker1 := sequence.NewChainWalker([]*sequence.ChainNode{{E: execHosts127}}, nil)
+ if err := s.Exec(context.Background(), qCtx1, walker1); err != nil {
+ t.Fatalf("Exec failed: %v", err)
+ }
+
+ if s.blockedQueries.Load() != 0 {
+ t.Errorf("expected blockedQueries 0 for hosts 127.0.0.1, got %d", s.blockedQueries.Load())
+ }
+ if s.hostsQueries.Load() != 1 {
+ t.Errorf("expected hostsQueries 1, got %d", s.hostsQueries.Load())
+ }
+ _, logs1 := s.ringBuffer.QueryLogs(1, 0, "", "all")
+ if len(logs1) != 1 || logs1[0].IsBlocked || logs1[0].Upstream != "hosts" {
+ t.Errorf("expected unblocked hosts log for 127.0.0.1, got %+v", logs1)
+ }
+
+ // 2. Hosts returning 127.0.0.1 for an ad/blocked domain (e.g. baidu.com) -> IS blocked
+ q2 := new(dns.Msg)
+ q2.SetQuestion("baidu.com.", dns.TypeA)
+ qCtx2 := query_context.NewContext(q2)
+
+ execHostsBaidu127 := sequence.ExecutableFunc(func(ctx context.Context, qCtx *query_context.Context) error {
+ resp := new(dns.Msg)
+ resp.SetReply(qCtx.Q())
+ resp.Answer = append(resp.Answer, &dns.A{
+ Hdr: dns.RR_Header{Name: "baidu.com.", Rrtype: dns.TypeA, Class: dns.ClassINET, Ttl: 300},
+ A: []byte{127, 0, 0, 1},
+ })
+ qCtx.SetResponse(resp)
+ qCtx.SetFromHosts()
+ return nil
+ })
+
+ walker2 := sequence.NewChainWalker([]*sequence.ChainNode{{E: execHostsBaidu127}}, nil)
+ if err := s.Exec(context.Background(), qCtx2, walker2); err != nil {
+ t.Fatalf("Exec failed: %v", err)
+ }
+
+ if s.blockedQueries.Load() != 1 {
+ t.Errorf("expected blockedQueries 1 for hosts 127.0.0.1 baidu.com, got %d", s.blockedQueries.Load())
+ }
+ if s.hostsQueries.Load() != 1 {
+ t.Errorf("expected hostsQueries to remain 1, got %d", s.hostsQueries.Load())
+ }
+ _, logs2 := s.ringBuffer.QueryLogs(1, 0, "", "all")
+ if len(logs2) != 1 || !logs2[0].IsBlocked {
+ t.Errorf("expected blocked log for hosts 127.0.0.1 baidu.com, got %+v", logs2)
+ }
+
+ // 3. Local black_hole returning 127.0.0.1 (no upstream, no hosts) -> IS blocked
+ q3 := new(dns.Msg)
+ q3.SetQuestion("ad.blackhole.com.", dns.TypeA)
+ qCtx3 := query_context.NewContext(q3)
+
+ execBlackhole127 := sequence.ExecutableFunc(func(ctx context.Context, qCtx *query_context.Context) error {
+ resp := new(dns.Msg)
+ resp.SetReply(qCtx.Q())
+ resp.Answer = append(resp.Answer, &dns.A{
+ Hdr: dns.RR_Header{Name: "ad.blackhole.com.", Rrtype: dns.TypeA, Class: dns.ClassINET, Ttl: 300},
+ A: []byte{127, 0, 0, 1},
+ })
+ qCtx.SetResponse(resp)
+ return nil
+ })
+
+ walker3 := sequence.NewChainWalker([]*sequence.ChainNode{{E: execBlackhole127}}, nil)
+ if err := s.Exec(context.Background(), qCtx3, walker3); err != nil {
+ t.Fatalf("Exec failed: %v", err)
+ }
+
+ if s.blockedQueries.Load() != 2 {
+ t.Errorf("expected blockedQueries 2 for black_hole 127.0.0.1, got %d", s.blockedQueries.Load())
+ }
+ _, logs3 := s.ringBuffer.QueryLogs(1, 0, "", "all")
+ if len(logs3) != 1 || !logs3[0].IsBlocked || logs3[0].Rule != "block" {
+ t.Errorf("expected blocked log with rule 'block', got %+v", logs3)
+ }
+}
+
+func TestStatsAPIQueryLogsExtendedFilters(t *testing.T) {
+ rb := NewRingBuffer(10)
+ rb.Push(LogEntry{Domain: "t.com.", Status: "TIMEOUT", Upstream: "upstream1"})
+ rb.Push(LogEntry{Domain: "e.com.", Status: "ERROR", Upstream: "upstream1"})
+ rb.Push(LogEntry{Domain: "nx.com.", Status: "NXDOMAIN", Upstream: "upstream1"})
+ rb.Push(LogEntry{Domain: "h.com.", Status: "NOERROR", Upstream: "hosts", Rule: "hosts"})
+
+ totalTimeout, logsTimeout := rb.QueryLogs(10, 0, "", "timeout")
+ if totalTimeout != 1 || logsTimeout[0].Domain != "t.com." {
+ t.Errorf("filter timeout failed: total=%d, logs=%+v", totalTimeout, logsTimeout)
+ }
+
+ totalError, logsError := rb.QueryLogs(10, 0, "", "error")
+ if totalError != 1 || logsError[0].Domain != "e.com." {
+ t.Errorf("filter error failed: total=%d, logs=%+v", totalError, logsError)
+ }
+
+ totalNX, logsNX := rb.QueryLogs(10, 0, "", "nxdomain")
+ if totalNX != 1 || logsNX[0].Domain != "nx.com." {
+ t.Errorf("filter nxdomain failed: total=%d, logs=%+v", totalNX, logsNX)
+ }
+
+ totalHosts, logsHosts := rb.QueryLogs(10, 0, "", "hosts")
+ if totalHosts != 1 || logsHosts[0].Domain != "h.com." {
+ t.Errorf("filter hosts failed: total=%d, logs=%+v", totalHosts, logsHosts)
+ }
+}
+
+func TestHistoryStats168Hours(t *testing.T) {
+ h := NewHistoryStats()
+ now := time.Now()
+
+ // Record point 100 hours ago
+ h.Record(now.Add(-100*time.Hour), false, true)
+
+ points := h.GetHistory(168)
+ if len(points) != 168 {
+ t.Fatalf("expected 168 points, got %d", len(points))
+ }
+
+ point100 := points[len(points)-1-100]
+ if point100.Total != 1 || point100.Cached != 1 {
+ t.Errorf("expected 1 total query 100 hours ago, got %+v", point100)
+ }
+}
@@ -0,0 +1,207 @@
From 21c024d30bba7603f5b002fc235b0cad11c72a4a Mon Sep 17 00:00:00 2001
From: sbwml <admin@cooluc.com>
Date: Thu, 15 Oct 2026 19:12:07 +0800
Subject: [PATCH 5/5] feat(stats_api): isolate timeout queries from average
latency calculation
- Exclude timeout request latencies from totalLatencyUs to prevent skewing average latency
- Add timeoutQueries and latencyQueries atomic counters
- Expose timeout_queries and timeout_percentage in /api/v1/stats API
- Persist timeout and latency query counters in stats dump with backward compatibility
- Add unit test verifying latency calculation purity under timeout conditions
---
plugin/executable/stats_api/stats_api.go | 35 ++++++++-
plugin/executable/stats_api/stats_api_test.go | 72 +++++++++++++++++++
2 files changed, 104 insertions(+), 3 deletions(-)
--- a/plugin/executable/stats_api/stats_api.go
+++ b/plugin/executable/stats_api/stats_api.go
@@ -556,7 +556,9 @@ type StatsDumpData struct {
CachedQueries uint64 `json:"cached_queries"`
HostsQueries uint64 `json:"hosts_queries,omitempty"`
ArbitraryQueries uint64 `json:"arbitrary_queries,omitempty"`
+ TimeoutQueries uint64 `json:"timeout_queries,omitempty"`
TotalLatencyUs uint64 `json:"total_latency_us"`
+ LatencyQueries uint64 `json:"latency_queries,omitempty"`
TopDomains map[string]uint64 `json:"top_domains,omitempty"`
TopClients map[string]uint64 `json:"top_clients,omitempty"`
TopBlocked map[string]uint64 `json:"top_blocked,omitempty"`
@@ -580,7 +582,9 @@ type StatsAPI struct {
cachedQueries atomic.Uint64
hostsQueries atomic.Uint64
arbitraryQueries atomic.Uint64
+ timeoutQueries atomic.Uint64
totalLatencyUs atomic.Uint64
+ latencyQueries atomic.Uint64
currentQPS atomic.Uint64
updatedCount atomic.Uint64
@@ -711,15 +715,23 @@ func (s *StatsAPI) handleStats(w http.Re
cached := s.cachedQueries.Load()
hosts := s.hostsQueries.Load()
arbitrary := s.arbitraryQueries.Load()
+ timeout := s.timeoutQueries.Load()
latUs := s.totalLatencyUs.Load()
+ latQueries := s.latencyQueries.Load()
- var blockedPct, cachedPct, avgLat float64
+ var blockedPct, cachedPct, timeoutPct, avgLat float64
if total > 0 {
blockedPct = float64(blocked) / float64(total) * 100.0
if blockedPct > 100.0 {
blockedPct = 100.0
}
- avgLat = (float64(latUs) / float64(total)) / 1000.0
+ timeoutPct = float64(timeout) / float64(total) * 100.0
+ if timeoutPct > 100.0 {
+ timeoutPct = 100.0
+ }
+ }
+ if latQueries > 0 {
+ avgLat = (float64(latUs) / float64(latQueries)) / 1000.0
}
excluded := blocked + hosts + arbitrary
if total > excluded {
@@ -731,6 +743,7 @@ func (s *StatsAPI) handleStats(w http.Re
blockedPct = math.Round(blockedPct*100) / 100
cachedPct = math.Round(cachedPct*100) / 100
+ timeoutPct = math.Round(timeoutPct*100) / 100
avgLat = math.Round(avgLat*100) / 100
qps := s.currentQPS.Load()
@@ -740,8 +753,10 @@ func (s *StatsAPI) handleStats(w http.Re
"cached_queries": cached,
"hosts_queries": hosts,
"arbitrary_queries": arbitrary,
+ "timeout_queries": timeout,
"blocked_percentage": blockedPct,
"cached_percentage": cachedPct,
+ "timeout_percentage": timeoutPct,
"avg_latency_ms": avgLat,
"qps": qps,
}
@@ -880,7 +895,9 @@ func (s *StatsAPI) writeDump(w io.Writer
CachedQueries: s.cachedQueries.Load(),
HostsQueries: s.hostsQueries.Load(),
ArbitraryQueries: s.arbitraryQueries.Load(),
+ TimeoutQueries: s.timeoutQueries.Load(),
TotalLatencyUs: s.totalLatencyUs.Load(),
+ LatencyQueries: s.latencyQueries.Load(),
}
data.TopDomains, data.TopClients, data.TopBlocked = s.topStats.Export()
@@ -916,7 +933,13 @@ func (s *StatsAPI) readDump(r io.Reader)
s.cachedQueries.Store(data.CachedQueries)
s.hostsQueries.Store(data.HostsQueries)
s.arbitraryQueries.Store(data.ArbitraryQueries)
+ s.timeoutQueries.Store(data.TimeoutQueries)
s.totalLatencyUs.Store(data.TotalLatencyUs)
+ latQueries := data.LatencyQueries
+ if latQueries == 0 && data.TotalQueries > data.TimeoutQueries {
+ latQueries = data.TotalQueries - data.TimeoutQueries
+ }
+ s.latencyQueries.Store(latQueries)
s.topStats.Import(data.TopDomains, data.TopClients, data.TopBlocked)
s.historyStats.Import(data.History)
@@ -1068,7 +1091,6 @@ func (s *StatsAPI) Exec(ctx context.Cont
elapsed := time.Since(start)
s.totalQueries.Add(1)
- s.totalLatencyUs.Add(uint64(elapsed.Microseconds()))
s.updatedCount.Add(1)
var clientIP string
@@ -1183,6 +1205,13 @@ func (s *StatsAPI) Exec(ctx context.Cont
s.blockedQueries.Add(1)
}
+ if isTimeout {
+ s.timeoutQueries.Add(1)
+ } else {
+ s.totalLatencyUs.Add(uint64(elapsed.Microseconds()))
+ s.latencyQueries.Add(1)
+ }
+
isHosts := qCtx.FromHosts && !isBlocked && !isCached && qCtx.UpstreamSelected == nil
if isHosts {
s.hostsQueries.Add(1)
--- a/plugin/executable/stats_api/stats_api_test.go
+++ b/plugin/executable/stats_api/stats_api_test.go
@@ -1063,3 +1063,75 @@ func TestHistoryStats168Hours(t *testing
t.Errorf("expected 1 total query 100 hours ago, got %+v", point100)
}
}
+
+func TestStatsAPITimeoutExclusionFromLatency(t *testing.T) {
+ s := NewStatsAPI(&Args{Capacity: 10}, zap.NewNop())
+
+ // 1. Successful fast query (e.g. 5ms)
+ q1 := new(dns.Msg)
+ q1.SetQuestion("fast.com.", dns.TypeA)
+ qCtx1 := query_context.NewContext(q1)
+
+ execFast := sequence.ExecutableFunc(func(ctx context.Context, qCtx *query_context.Context) error {
+ time.Sleep(5 * time.Millisecond)
+ resp := new(dns.Msg)
+ resp.SetReply(qCtx.Q())
+ resp.Answer = append(resp.Answer, &dns.A{
+ Hdr: dns.RR_Header{Name: "fast.com.", Rrtype: dns.TypeA, Class: dns.ClassINET, Ttl: 300},
+ A: []byte{1, 1, 1, 1},
+ })
+ qCtx.SetResponse(resp)
+ return nil
+ })
+
+ walker1 := sequence.NewChainWalker([]*sequence.ChainNode{{E: execFast}}, nil)
+ if err := s.Exec(context.Background(), qCtx1, walker1); err != nil {
+ t.Fatalf("fast query failed: %v", err)
+ }
+
+ // 2. Timeout query (simulated timeout with deadline exceeded error)
+ q2 := new(dns.Msg)
+ q2.SetQuestion("timeout.com.", dns.TypeA)
+ qCtx2 := query_context.NewContext(q2)
+
+ execTimeout := sequence.ExecutableFunc(func(ctx context.Context, qCtx *query_context.Context) error {
+ time.Sleep(50 * time.Millisecond)
+ return context.DeadlineExceeded
+ })
+
+ walker2 := sequence.NewChainWalker([]*sequence.ChainNode{{E: execTimeout}}, nil)
+ _ = s.Exec(context.Background(), qCtx2, walker2)
+
+ if s.totalQueries.Load() != 2 {
+ t.Errorf("expected 2 total queries, got %d", s.totalQueries.Load())
+ }
+ if s.timeoutQueries.Load() != 1 {
+ t.Errorf("expected 1 timeout query, got %d", s.timeoutQueries.Load())
+ }
+ if s.latencyQueries.Load() != 1 {
+ t.Errorf("expected 1 latency query (excluding timeout), got %d", s.latencyQueries.Load())
+ }
+
+ // Test HTTP /api/v1/stats output
+ req := httptest.NewRequest(http.MethodGet, "/api/v1/stats", nil)
+ w := httptest.NewRecorder()
+ s.Router().ServeHTTP(w, req)
+
+ var stats map[string]any
+ if err := json.Unmarshal(w.Body.Bytes(), &stats); err != nil {
+ t.Fatalf("failed to parse stats response: %v", err)
+ }
+
+ if stats["timeout_queries"].(float64) != 1 {
+ t.Errorf("expected timeout_queries == 1, got %v", stats["timeout_queries"])
+ }
+ if stats["timeout_percentage"].(float64) != 50.0 {
+ t.Errorf("expected timeout_percentage == 50.0, got %v", stats["timeout_percentage"])
+ }
+
+ // avg_latency_ms should be around 5ms, definitely not pulled up to >20ms by the 50ms timeout!
+ avgLat := stats["avg_latency_ms"].(float64)
+ if avgLat >= 20.0 {
+ t.Errorf("avg_latency_ms %v ms was polluted by timeout query!", avgLat)
+ }
+}