diff --git a/Makefile b/Makefile index d27e516..da2db7b 100644 --- a/Makefile +++ b/Makefile @@ -63,7 +63,16 @@ web-dev: ## Run the Vite dev server (proxies /api to 127.0.0.1:8080) ## Verify .PHONY: check -check: fmt-check vet test ## What CI runs +check: check-go test ## What CI runs (format, vet, compile, then the suite) + +.PHONY: check-go +check-go: fmt-check vet go-build ## The tight inner loop: format, vet, compile. No npm, no tests + @echo "check-go: ok" + +.PHONY: gofmt-w +gofmt-w: ## Format every Go file in the tree, including ones just created + @gofmt -w $(GOFILES) + @echo "gofmt: ok" .PHONY: test test: ## Run the test suite with the race detector $(GO) test $(TESTFLAGS) ./... diff --git a/internal/qlog/qlog.go b/internal/qlog/qlog.go index 4f14b8d..d63cc0f 100644 --- a/internal/qlog/qlog.go +++ b/internal/qlog/qlog.go @@ -93,9 +93,9 @@ type Logger struct { // live is a fixed-size ring of the most recent entries. Each slot is a // pointer, so a writer on the query path never races a reader walking the // ring: publishing is one atomic word, and a reader sees either the whole - // entry or the one it replaced. + // entry or the one it replaced. The slot an entry lands in comes from its + // id, so concurrent writers cannot leave a hole or reorder the stream. live []atomic.Pointer[liveEntry] - livePos atomic.Uint32 liveMu sync.RWMutex liveCond *sync.Cond liveFull atomic.Bool @@ -104,6 +104,10 @@ type Logger struct { pendingMu sync.Mutex pending []Entry + // closed is set by Close; queries can still be logged after shutdown, so + // Log has to notice rather than write to a closed bbolt handle. + closed atomic.Bool + // counts are rolling per-domain counters. counts *domainCounts @@ -208,8 +212,20 @@ func New(opt Options) (*Logger, error) { if err != nil { return err } - _, err = tx.CreateBucketIfNotExists(bucketMeta) - return err + meta, err := tx.CreateBucketIfNotExists(bucketMeta) + if err != nil { + return err + } + // Resume the id sequence where the previous process stopped. The + // bbolt keys are that same sequence, so ids stay unique across a + // restart: a repeated id would make the SSE stream and the UI show + // a live entry as an older one carrying the same id. + if v := meta.Get(keyNextID); len(v) == 8 { + // next_id is the id the next entry will take, while seq + // counts ids already handed out. + l.seq.Store(beUint64(v) - 1) + } + return nil }); err != nil { _ = db.Close() return nil, err @@ -251,6 +267,9 @@ func (l *Logger) ShouldLog(a Action) bool { // Log records a query. It never blocks on bbolt: the entry goes to the live // ring and a batch, and a background goroutine does the disk write. func (l *Logger) Log(e Entry) { + if l == nil || l.closed.Load() { + return + } if e.TS.IsZero() { e.TS = l.opts.Now() } @@ -264,8 +283,9 @@ func (l *Logger) Log(e Entry) { // Live ring. The slot is published with a single atomic store; a reader // walking the ring concurrently gets either the old or the new entry, // never a half-written one. - pos := l.livePos.Add(1) - 1 - l.live[pos%uint32(len(l.live))].Store(&liveEntry{entry: e}) + // The index is the entry id rather than a second counter, so a burst from + // several resolver goroutines still lands gap free and in id order. + l.live[(e.ID-1)%uint64(len(l.live))].Store(&liveEntry{entry: e}) // Fan out to SSE subscribers (non-blocking: a slow consumer drops entries // rather than stalling the query path). @@ -308,7 +328,7 @@ func (l *Logger) flushLoop() { // Flush writes any pending entries to bbolt and trims the ring. func (l *Logger) Flush() { - if l.db == nil { + if l == nil || l.db == nil { return } l.pendingMu.Lock() @@ -359,17 +379,34 @@ func (l *Logger) trim(b *bolt.Bucket, nextID uint64) error { if l.opts.MaxEntries <= 0 { return nil } - stats := b.Stats() - if stats.KeyN <= l.opts.MaxEntries { + // The keys are a dense id range, so the count is the span from the oldest + // surviving key up to nextID. Bucket.Stats cannot answer that here: it + // walks the pages as they sit on disk, and a write transaction has not + // spilled this batch yet, so it always under-counts and the cap is never + // enforced. + cur := b.Cursor() + first, _ := cur.First() + if first == nil { return nil } - excess := stats.KeyN - l.opts.MaxEntries - cur := b.Cursor() - for k, _ := cur.First(); k != nil && excess > 0; k, _ = cur.Next() { + oldest := beUint64(first) + span := nextID - oldest + if span <= uint64(l.opts.MaxEntries) { + return nil + } + excess := span - uint64(l.opts.MaxEntries) + // Collect the doomed keys before deleting any: bbolt rewrites a leaf + // page in place once it drops under a quarter full, which invalidates the + // cursor's page stack, so deleting while iterating stops early and the + // cap is only half enforced. + keys := make([][]byte, 0, excess) + for k, _ := cur.First(); k != nil && len(keys) < int(excess); k, _ = cur.Next() { + keys = append(keys, append([]byte(nil), k...)) + } + for _, k := range keys { if err := b.Delete(k); err != nil { return err } - excess-- } return nil } @@ -426,6 +463,9 @@ type Page struct { // It powers the SSE initial burst so a client connecting mid-stream sees // recent history immediately. func (l *Logger) Recent(limit int) []Entry { + if l == nil { + return nil + } if limit <= 0 { limit = 200 } @@ -433,14 +473,14 @@ func (l *Logger) Recent(limit int) []Entry { limit = len(l.live) } out := make([]Entry, 0, limit) - pos := l.livePos.Load() + size := uint64(len(l.live)) + pos := l.seq.Load() // the newest id doubles as the newest ring position total := pos - if total > uint32(len(l.live)) { - total = uint32(len(l.live)) + if total > size { + total = size } - for i := uint32(0); i < total; i++ { - idx := (pos - 1 - i) % uint32(len(l.live)) - le := l.live[idx].Load() + for i := uint64(0); i < total; i++ { + le := l.live[(pos-1-i)%size].Load() if le == nil { // Never written (or already aged out of the window). continue @@ -455,6 +495,9 @@ func (l *Logger) Recent(limit int) []Entry { // Search queries the persisted log. func (l *Logger) Search(q Query) (Page, error) { + if l == nil { + return Page{Offset: q.Offset, Limit: q.Limit}, nil + } l.Flush() page := Page{Offset: q.Offset, Limit: q.Limit} if q.Limit <= 0 { @@ -623,14 +666,22 @@ func (l *Logger) SubscriberCount() int { // Clear removes every persisted entry. func (l *Logger) Clear() error { - if l.db == nil { + if l == nil || l.db == nil { return nil } err := l.db.Update(func(tx *bolt.Tx) error { if tx.Bucket(bucketLog) == nil { - return nil + _, err := tx.CreateBucket(bucketLog) + return err } - return tx.DeleteBucket(bucketLog) + if err := tx.DeleteBucket(bucketLog); err != nil { + return err + } + // Recreate it in the same transaction: the flush path assumes the + // bucket is there, and a logger that silently stopped persisting after + // a clear would look like a healthy, empty log. + _, err := tx.CreateBucket(bucketLog) + return err }) if err != nil { l.logErrors.Add(1) @@ -640,6 +691,9 @@ func (l *Logger) Clear() error { // Count is the number of persisted entries. func (l *Logger) Count() int { + if l == nil { + return 0 + } if l.db == nil { return int(l.seq.Load()) } @@ -708,6 +762,10 @@ func (l *Logger) Stats() Stats { // Close flushes and releases resources. func (l *Logger) Close() error { + if l == nil { + return nil + } + l.closed.Store(true) select { case <-l.stop: default: diff --git a/internal/qlog/qlog_test.go b/internal/qlog/qlog_test.go new file mode 100644 index 0000000..b0db0ee --- /dev/null +++ b/internal/qlog/qlog_test.go @@ -0,0 +1,838 @@ +package qlog + +import ( + "fmt" + "path/filepath" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/miekg/dns" +) + +// openLogger builds a Logger over a temp-dir bbolt file. Every logger this +// file creates is closed through t.Cleanup, so a failing assertion cannot leak +// the flush goroutine or hold the bbolt file lock. +func openLogger(t *testing.T, path string, mutate func(*Options)) *Logger { + t.Helper() + opt := Options{ + Path: path, + MaxEntries: 1000, + LiveBuffer: 256, + FlushInterval: 20 * time.Millisecond, + SampleRate: 1, + LogBlocked: true, + LogCached: true, + LogErrors: true, + TopDomains: 128, + TopTTL: time.Minute, + } + if mutate != nil { + mutate(&opt) + } + l, err := New(opt) + if err != nil { + t.Fatalf("qlog.New: %v", err) + } + t.Cleanup(func() { _ = l.Close() }) + return l +} + +func tempPath(t *testing.T) string { + t.Helper() + return filepath.Join(t.TempDir(), "queries.db") +} + +// entryFor builds an entry the way the engine does: the question-derived +// fields come from a real dns.Msg via FromMsg, the rest is filled in by the +// caller. +func entryFor(i int) Entry { + q := new(dns.Msg) + q.SetQuestion(fmt.Sprintf("host%d.test.", i), dns.TypeA) + m := new(dns.Msg) + m.SetReply(q) + rr, err := dns.NewRR(q.Question[0].Name + " 30 IN A 203.0.113.7") + if err != nil { + panic(err) + } + m.Answer = []dns.RR{rr} + e := Entry{ + TS: time.Date(2024, 5, 1, 12, 0, i, 0, time.UTC), + Client: "192.0.2.10", + Group: "testers", + Duration: int64(i) * 2, + Action: ActionAllow, + Upstream: "127.0.0.1:5399", + List: "test-list", + Protocol: "udp", + } + e.FromMsg(m) + return e +} + +func names(items []Entry) []string { + out := make([]string, 0, len(items)) + for _, e := range items { + out = append(out, e.Name) + } + return out +} + +// TestLogRoundTripReopen is the core promise of the package: what Log accepted +// is what Search hands back after the process restarted. +func TestLogRoundTripReopen(t *testing.T) { + path := tempPath(t) + first := openLogger(t, path, nil) + for i := 1; i <= 5; i++ { + first.Log(entryFor(i)) + } + first.Flush() + if got := first.Count(); got != 5 { + t.Fatalf("Count() = %d, want 5", got) + } + // Close before reopening: bbolt takes an exclusive file lock, so a + // reopened handle would just block on the timeout. + if err := first.Close(); err != nil { + t.Fatalf("close: %v", err) + } + + second := openLogger(t, path, nil) + page, err := second.Search(Query{Limit: 50}) + if err != nil { + t.Fatalf("search: %v", err) + } + if page.Total != 5 { + t.Fatalf("Total = %d, want 5", page.Total) + } + want := []string{"host5.test.", "host4.test.", "host3.test.", "host2.test.", "host1.test."} + if got := names(page.Items); !equalStrings(got, want) { + t.Fatalf("names = %v, want %v (newest first)", got, want) + } + for i, e := range page.Items { + w := entryFor(5 - i) + assertEntryFields(t, e, w) + } + if got := second.Count(); got != 5 { + t.Errorf("Count() after reopen = %d, want 5", got) + } + if n := second.ErrorCount(); n != 0 { + t.Errorf("ErrorCount() = %d, want 0", n) + } +} + +// TestReopenContinuesEntryIDs pins that ids stay unique across a restart: a +// logger that resumed at 1 would hand the UI two entries with the same id. +func TestReopenContinuesEntryIDs(t *testing.T) { + path := tempPath(t) + first := openLogger(t, path, nil) + for i := 1; i <= 3; i++ { + first.Log(entryFor(i)) + } + first.Flush() + if err := first.Close(); err != nil { + t.Fatalf("close: %v", err) + } + + second := openLogger(t, path, nil) + second.Log(entryFor(4)) + second.Flush() + page, err := second.Search(Query{Limit: 50}) + if err != nil { + t.Fatalf("search: %v", err) + } + if len(page.Items) != 4 { + t.Fatalf("items = %d, want 4", len(page.Items)) + } + seen := map[uint64]bool{} + for _, e := range page.Items { + if seen[e.ID] { + t.Fatalf("duplicate entry id %d in %v", e.ID, names(page.Items)) + } + seen[e.ID] = true + } + if page.Items[0].ID != 4 { + t.Errorf("newest id = %d, want 4 (ids must continue across a restart)", page.Items[0].ID) + } +} + +// assertEntryFields compares every field the round trip has to preserve. +func assertEntryFields(t *testing.T, got, want Entry) { + t.Helper() + if !got.TS.Equal(want.TS) { + t.Errorf("TS = %v, want %v", got.TS, want.TS) + } + if got.Client != want.Client { + t.Errorf("Client = %q, want %q", got.Client, want.Client) + } + if got.Group != want.Group { + t.Errorf("Group = %q, want %q", got.Group, want.Group) + } + if got.Name != want.Name { + t.Errorf("Name = %q, want %q", got.Name, want.Name) + } + if got.QType != want.QType { + t.Errorf("QType = %d, want %d", got.QType, want.QType) + } + if got.QTypeStr != want.QTypeStr { + t.Errorf("QTypeStr = %q, want %q", got.QTypeStr, want.QTypeStr) + } + if got.RCode != want.RCode { + t.Errorf("RCode = %d, want %d", got.RCode, want.RCode) + } + if got.Action != want.Action { + t.Errorf("Action = %q, want %q", got.Action, want.Action) + } + if got.Duration != want.Duration { + t.Errorf("Duration = %d, want %d", got.Duration, want.Duration) + } + if got.Upstream != want.Upstream { + t.Errorf("Upstream = %q, want %q", got.Upstream, want.Upstream) + } + if got.List != want.List { + t.Errorf("List = %q, want %q", got.List, want.List) + } + if got.Protocol != want.Protocol { + t.Errorf("Protocol = %q, want %q", got.Protocol, want.Protocol) + } + if got.Blocked != want.Blocked { + t.Errorf("Blocked = %v, want %v", got.Blocked, want.Blocked) + } + if got.RCodeName() != "NOERROR" { + t.Errorf("RCodeName() = %q, want NOERROR", got.RCodeName()) + } +} + +func equalStrings(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} + +// TestMaxEntriesEvictsOldestFirst covers the on-disk size bound: the ring +// never grows past MaxEntries and the survivors are the newest entries. +func TestMaxEntriesEvictsOldestFirst(t *testing.T) { + l := openLogger(t, tempPath(t), func(o *Options) { o.MaxEntries = 8 }) + for i := 1; i <= 20; i++ { + l.Log(entryFor(i)) + } + l.Flush() + if got := l.Count(); got > 8 { + t.Fatalf("Count() = %d, want <= 8", got) + } + if got := l.Count(); got != 8 { + t.Fatalf("Count() = %d, want 8 (the ring should be full, not short)", got) + } + page, err := l.Search(Query{Limit: 100}) + if err != nil { + t.Fatalf("search: %v", err) + } + want := []string{ + "host20.test.", "host19.test.", "host18.test.", "host17.test.", + "host16.test.", "host15.test.", "host14.test.", "host13.test.", + } + if got := names(page.Items); !equalStrings(got, want) { + t.Fatalf("survivors = %v, want %v", got, want) + } + // A second flush must not trim below the bound either: the key count is + // re-checked on every transaction. + l.Flush() + if got := l.Count(); got != 8 { + t.Errorf("Count() after re-flush = %d, want 8", got) + } +} + +// TestMaxEntriesZeroMeansUnbounded documents the escape hatch: 0 disables the +// trim entirely. +func TestMaxEntriesZeroMeansUnbounded(t *testing.T) { + l := openLogger(t, tempPath(t), func(o *Options) { o.MaxEntries = 0 }) + for i := 1; i <= 25; i++ { + l.Log(entryFor(i)) + } + l.Flush() + if got := l.Count(); got != 25 { + t.Fatalf("Count() = %d, want 25", got) + } +} + +// TestLiveRingOrderAndGapFree pins the SSE initial burst: newest first, one +// step at a time, with no holes where the ring wrapped. +func TestLiveRingOrderAndGapFree(t *testing.T) { + l := openLogger(t, tempPath(t), func(o *Options) { o.LiveBuffer = 32 }) + for i := 1; i <= 40; i++ { + l.Log(entryFor(i)) + } + got := l.Recent(32) + if len(got) != 32 { + t.Fatalf("Recent(32) returned %d entries, want 32", len(got)) + } + for i, e := range got { + want := 40 - i + if e.ID != uint64(want) { + t.Fatalf("entry %d has id %d, want %d (ring must be newest first, gap free)", i, e.ID, want) + } + if e.Name != fmt.Sprintf("host%d.test.", want) { + t.Fatalf("entry %d name = %q, want host%d.test.", i, e.Name, want) + } + } + if got := l.Recent(5); len(got) != 5 || got[0].ID != 40 || got[4].ID != 36 { + t.Fatalf("Recent(5) = %v, want ids 40..36", got) + } + // limit <= 0 falls back to the 200-entry default, bounded by the ring. + if got := l.Recent(0); len(got) != 32 { + t.Fatalf("Recent(0) returned %d entries, want the whole 32-slot ring", len(got)) + } + // A limit past the ring size clamps instead of panicking. + if got := l.Recent(10000); len(got) != 32 { + t.Fatalf("Recent(10000) returned %d entries, want 32", len(got)) + } +} + +// TestLiveRingEmptyBeforeAnyWrite covers the zero state. +func TestLiveRingEmptyBeforeAnyWrite(t *testing.T) { + l := openLogger(t, tempPath(t), nil) + if got := l.Recent(10); len(got) != 0 { + t.Fatalf("Recent(10) on an empty ring = %d entries, want 0", len(got)) + } + if got := l.Stats().TotalLogged; got != 0 { + t.Fatalf("TotalLogged = %d, want 0", got) + } +} + +// TestLiveRingConcurrentLogAndRecent is the race test for the whole point of +// the []atomic.Pointer[liveEntry] ring: many writers publish while readers +// walk, and no reader may ever see a half-written entry. Every entry carries +// its own writer id and sequence in two fields, so a torn read — the failure +// mode a mutex-free ring has to avoid — shows up as Name not matching +// Client+Protocol. +func TestLiveRingConcurrentLogAndRecent(t *testing.T) { + const ( + writers = 8 + perWriter = 200 + readers = 3 + ) + l := openLogger(t, tempPath(t), func(o *Options) { + o.LiveBuffer = 2048 + o.Path = "" // the live ring is what is under test; skip bbolt + }) + var start sync.WaitGroup + var done sync.WaitGroup + start.Add(1) + for w := 0; w < writers; w++ { + w := w + done.Add(1) + go func() { + defer done.Done() + start.Wait() + for i := 0; i < perWriter; i++ { + l.Log(Entry{ + Client: "w" + strconv.Itoa(w), + Name: fmt.Sprintf("w%d-i%d.test.", w, i), + Protocol: strconv.Itoa(i), + QType: dns.TypeA, + QTypeStr: "A", + Action: ActionAllow, + TS: time.Unix(int64(i), 0), + }) + } + }() + } + stop := make(chan struct{}) + var readersWG sync.WaitGroup + for r := 0; r < readers; r++ { + readersWG.Add(1) + go func() { + defer readersWG.Done() + start.Wait() + for { + select { + case <-stop: + return + default: + } + snapshot := l.Recent(64) + seenID := make(map[uint64]bool, len(snapshot)) + seenKey := make(map[string]bool, len(snapshot)) + for _, e := range snapshot { + // Name and the client/protocol pair are written together + // into one immutable entry, so a reader that saw a mix + // of two would be reading torn data. + want := fmt.Sprintf("%s-i%s.test.", e.Client, e.Protocol) + if e.Name != want { + t.Errorf("torn entry: name %q does not match client %q/protocol %q", e.Name, e.Client, e.Protocol) + return + } + if e.ID == 0 { + t.Errorf("entry with id 0 reached a reader: %+v", e) + return + } + if e.ID > writers*perWriter { + t.Errorf("entry id %d is beyond the %d written", e.ID, writers*perWriter) + return + } + if seenID[e.ID] { + t.Errorf("id %d appeared twice in one snapshot", e.ID) + return + } + seenID[e.ID] = true + key := e.Client + "/" + e.Protocol + if seenKey[key] { + t.Errorf("entry w%s/%s appeared twice in one snapshot", e.Client, e.Protocol) + return + } + seenKey[key] = true + } + } + }() + } + start.Done() + done.Wait() + close(stop) + readersWG.Wait() + + // With the writers done the ring must be a complete, ordered history. + all := l.Recent(writers * perWriter) + if len(all) != writers*perWriter { + t.Fatalf("Recent returned %d entries, want %d", len(all), writers*perWriter) + } + for i, e := range all { + want := writers*perWriter - i + if e.ID != uint64(want) { + t.Fatalf("entry %d has id %d, want %d (final snapshot must be gap free)", i, e.ID, want) + } + } +} + +// TestSSEFanoutDeliversEachEntryOnce covers the live stream: every subscriber +// sees the same entries, in order, exactly once. +func TestSSEFanoutDeliversEachEntryOnce(t *testing.T) { + l := openLogger(t, tempPath(t), func(o *Options) { o.Path = "" }) + const subs = 3 + chans := make([]<-chan Entry, 0, subs) + cancels := make([]func(), 0, subs) + for i := 0; i < subs; i++ { + ch, cancel := l.Subscribe() + chans = append(chans, ch) + cancels = append(cancels, cancel) + } + if got := l.SubscriberCount(); got != subs { + t.Fatalf("SubscriberCount() = %d, want %d", got, subs) + } + for i := 1; i <= 5; i++ { + l.Log(entryFor(i)) + } + for s, ch := range chans { + for i := 1; i <= 5; i++ { + select { + case e := <-ch: + if e.ID != uint64(i) { + t.Errorf("subscriber %d: entry %d has id %d", s, i, e.ID) + } + if e.Name != fmt.Sprintf("host%d.test.", i) { + t.Errorf("subscriber %d: entry %d name = %q", s, i, e.Name) + } + default: + t.Fatalf("subscriber %d never got entry %d", s, i) + } + } + select { + case extra := <-ch: + t.Errorf("subscriber %d got an extra entry: %+v", s, extra) + default: + } + } + for _, cancel := range cancels { + cancel() + } +} + +// TestSSESlowSubscriberDoesNotBlockLog is the reason fanout is non-blocking: a +// consumer that stopped reading must never stall the query path. +func TestSSESlowSubscriberDoesNotBlockLog(t *testing.T) { + l := openLogger(t, tempPath(t), func(o *Options) { o.Path = "" }) + slow, cancelSlow := l.Subscribe() + defer cancelSlow() + // Nothing ever reads from slow, so its 256-slot buffer fills and stays + // full. + for i := 0; i < 256; i++ { + l.Log(entryFor(i + 1)) + } + done := make(chan struct{}) + go func() { + defer close(done) + for i := 0; i < 50; i++ { + l.Log(entryFor(i + 1)) + } + }() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("Log blocked on a subscriber that stopped reading") + } + // A healthy subscriber still gets everything. + fast, cancelFast := l.Subscribe() + defer cancelFast() + l.Log(entryFor(999)) + select { + case e := <-fast: + if e.Name != "host999.test." { + t.Errorf("fast subscriber got %q", e.Name) + } + case <-time.After(5 * time.Second): + t.Fatal("fast subscriber did not receive the entry") + } + // The slow consumer's buffer only ever holds entries from the first + // burst: everything logged after it filled up was dropped, not queued. + drained := 0 + for { + select { + case e := <-slow: + if e.ID > 256 { + t.Fatalf("slow subscriber received a dropped-then-queued entry: %+v", e) + } + drained++ + continue + default: + } + break + } + if drained != 256 { + t.Errorf("slow subscriber buffered %d entries, want 256", drained) + } + select { + case e := <-slow: + t.Errorf("slow subscriber received a late entry: %+v", e) + default: + } +} + +// TestSSEUnsubscribeStopsDelivery covers teardown: no delivery after the +// cancel func, a closed channel, and no leaked subscription slot. +func TestSSEUnsubscribeStopsDelivery(t *testing.T) { + l := openLogger(t, tempPath(t), func(o *Options) { o.Path = "" }) + ch, cancel := l.Subscribe() + l.Log(entryFor(1)) + if e := <-ch; e.ID != 1 { + t.Fatalf("first entry id = %d, want 1", e.ID) + } + cancel() + if got := l.SubscriberCount(); got != 0 { + t.Fatalf("SubscriberCount() = %d after cancel, want 0", got) + } + l.Log(entryFor(2)) + // A closed channel yields immediately, so a plain receive is enough. + select { + case e, ok := <-ch: + if ok { + t.Fatalf("received %+v after unsubscribe", e) + } + case <-time.After(time.Second): + t.Fatal("channel was not closed by the cancel func") + } + // Cancelling twice must not panic on a double close. + cancel() + if got := l.SubscriberCount(); got != 0 { + t.Errorf("SubscriberCount() = %d, want 0", got) + } +} + +// TestFieldPreservationThroughMsg checks the question-derived fields come from +// the message and the flags are mirrored. +func TestFieldPreservationThroughMsg(t *testing.T) { + l := openLogger(t, tempPath(t), nil) + q := new(dns.Msg) + q.SetQuestion("Blocked.Example.COM.", dns.TypeAAAA) + m := new(dns.Msg) + m.SetReply(q) + m.Rcode = dns.RcodeNameError + e := Entry{ + TS: time.Date(2024, 5, 1, 12, 0, 0, 0, time.UTC), + Client: "2001:db8::5", + Group: "laptops", + Duration: 7, + Action: ActionBlock, + List: "adlist", + Protocol: "doh", + Upstream: "", + } + e.FromMsg(m) + l.Log(e) + l.Flush() + + page, err := l.Search(Query{Limit: 10}) + if err != nil { + t.Fatalf("search: %v", err) + } + if len(page.Items) != 1 { + t.Fatalf("items = %d, want 1", len(page.Items)) + } + got := page.Items[0] + if got.Name != "Blocked.Example.COM." { + t.Errorf("Name = %q, want the name verbatim", got.Name) + } + if got.QType != dns.TypeAAAA || got.QTypeStr != "AAAA" { + t.Errorf("QType = %d/%q, want AAAA/AAAA", got.QType, got.QTypeStr) + } + if got.RCode != dns.RcodeNameError || got.RCodeName() != "NXDOMAIN" { + t.Errorf("RCode = %d/%q, want NXDOMAIN", got.RCode, got.RCodeName()) + } + if !got.Blocked { + t.Error("Blocked = false, want the mirror of ActionBlock") + } + if got.Client != "2001:db8::5" || got.Group != "laptops" || got.Duration != 7 { + t.Errorf("client/group/duration = %q/%q/%d", got.Client, got.Group, got.Duration) + } + if got.List != "adlist" || got.Protocol != "doh" { + t.Errorf("list/protocol = %q/%q", got.List, got.Protocol) + } + if got.Upstream != "" { + t.Errorf("Upstream = %q, want empty (a blocked query never reaches one)", got.Upstream) + } + if got.TS.IsZero() { + t.Error("TS was dropped") + } + // A blocked query is the mirror's only source, so a non-block must clear it. + l.Log(entryFor(9)) + l.Flush() + page, err = l.Search(Query{BlockedOnly: true, Limit: 10}) + if err != nil { + t.Fatalf("search: %v", err) + } + if page.Total != 1 || page.Items[0].Name != "Blocked.Example.COM." { + t.Fatalf("BlockedOnly search = %+v, want just the blocked name", page.Items) + } +} + +// TestQueryStringStampsMissingFields covers the Log-time fixups: an entry +// built by hand gets a timestamp and a type name without the caller setting +// them. +func TestQueryStringStampsMissingFields(t *testing.T) { + l := openLogger(t, tempPath(t), func(o *Options) { o.Path = "" }) + l.Log(Entry{Name: "stamped.test.", QType: dns.TypeMX, Action: ActionAllow}) + got := l.Recent(1) + if len(got) != 1 { + t.Fatalf("Recent(1) = %d entries, want 1", len(got)) + } + if got[0].QTypeStr != "MX" { + t.Errorf("QTypeStr = %q, want MX", got[0].QTypeStr) + } + if got[0].TS.IsZero() { + t.Error("TS = zero, want the logger's clock") + } + if got[0].ID != 1 { + t.Errorf("ID = %d, want 1", got[0].ID) + } +} + +// TestDisabledLoggerStillServesTheLiveRing covers the no-path configuration: +// nothing is persisted, but the SSE ring, the counters and the API filters +// still work. +func TestDisabledLoggerStillServesTheLiveRing(t *testing.T) { + l := openLogger(t, tempPath(t), func(o *Options) { o.Path = "" }) + if l.Stats().Enabled { + t.Error("Stats().Enabled = true without a path") + } + for i := 1; i <= 3; i++ { + e := entryFor(i) + e.Action = ActionBlock + l.Log(e) + } + if got := l.Count(); got != 3 { + t.Errorf("Count() = %d, want the 3 logged ids", got) + } + page, err := l.Search(Query{Action: "block", Limit: 10}) + if err != nil { + t.Fatalf("search: %v", err) + } + if page.Total != 3 { + t.Fatalf("Total = %d, want 3 (search must fall back to the live ring)", page.Total) + } + // Offset and text filters work on the same fallback path. + page, err = l.Search(Query{Text: "host2", Limit: 10}) + if err != nil { + t.Fatalf("search: %v", err) + } + if page.Total != 1 || page.Items[0].Name != "host2.test." { + t.Fatalf("text search = %+v, want host2.test.", page.Items) + } + if err := l.Clear(); err != nil { + t.Errorf("Clear() without persistence: %v", err) + } + if err := l.Close(); err != nil { + t.Errorf("Close() without persistence: %v", err) + } +} + +// TestLogOnClosedLoggerIsANoOp covers the shutdown path: every read and write +// after Close must stay safe rather than panicking on a closed bbolt handle. +func TestLogOnClosedLoggerIsANoOp(t *testing.T) { + l := openLogger(t, tempPath(t), nil) + l.Log(entryFor(1)) + if err := l.Close(); err != nil { + t.Fatalf("close: %v", err) + } + l.Log(entryFor(2)) + l.Flush() + _ = l.Count() + _ = l.Recent(10) + if _, err := l.Search(Query{Limit: 10}); err == nil { + t.Log("search on a closed store returned no error; acceptable") + } + if got := l.ErrorCount(); got > 1 { + t.Errorf("ErrorCount() = %d, want at most the failed write", got) + } + // A nil logger is the "logging disabled" shape some callers hold; it must + // swallow the call instead of panicking on the query path. + var nilLogger *Logger + nilLogger.Log(entryFor(3)) + if got := nilLogger.Recent(1); got != nil { + t.Errorf("nil Recent() = %v, want nil", got) + } +} + +// TestShouldLogPolicy covers the sampling gate the engine consults. +func TestShouldLogPolicy(t *testing.T) { + all := openLogger(t, tempPath(t), func(o *Options) { o.Path = "" }) + for _, a := range []Action{ActionAllow, ActionBlock, ActionCached, ActionError, ActionDrop, ActionRefused} { + if !all.ShouldLog(a) { + t.Errorf("ShouldLog(%q) = false with every category enabled", a) + } + } + off := openLogger(t, tempPath(t), func(o *Options) { + o.Path = "" + o.LogBlocked = false + o.LogCached = false + o.LogErrors = false + }) + if off.ShouldLog(ActionBlock) || off.ShouldLog(ActionCached) || off.ShouldLog(ActionError) || off.ShouldLog(ActionDrop) { + t.Error("disabled categories must not be logged") + } + if !off.ShouldLog(ActionAllow) { + t.Error("ordinary resolutions still pass a full sample rate") + } +} + +// TestClearEmptiesPersistedLog covers the admin API's "clear log" button. +func TestClearEmptiesPersistedLog(t *testing.T) { + l := openLogger(t, tempPath(t), nil) + for i := 1; i <= 4; i++ { + l.Log(entryFor(i)) + } + l.Flush() + if err := l.Clear(); err != nil { + t.Fatalf("clear: %v", err) + } + if got := l.Count(); got != 0 { + t.Errorf("Count() = %d after Clear, want 0", got) + } + page, err := l.Search(Query{Limit: 10}) + if err != nil { + t.Fatalf("search: %v", err) + } + if page.Total != 0 { + t.Errorf("Total = %d after Clear, want 0", page.Total) + } + // Logging keeps working afterwards, and the new entries are found. + l.Log(entryFor(9)) + l.Flush() + page, err = l.Search(Query{Limit: 10}) + if err != nil { + t.Fatalf("search: %v", err) + } + if page.Total != 1 || page.Items[0].Name != "host9.test." { + t.Fatalf("search after clear = %+v, want host9.test.", page.Items) + } +} + +// TestTopDomainsCounters covers the rolling per-domain panel: blocked counts +// are tracked separately from the total. +func TestTopDomainsCounters(t *testing.T) { + l := openLogger(t, tempPath(t), func(o *Options) { o.Path = "" }) + for i := 0; i < 3; i++ { + e := entryFor(1) + e.Name = "ads.test." + e.Action = ActionBlock + l.Log(e) + } + for i := 0; i < 2; i++ { + l.Log(entryFor(1)) + } + blocked := l.TopBlocked(5) + if len(blocked) != 1 { + t.Fatalf("TopBlocked = %+v, want one domain", blocked) + } + if blocked[0].Name != "ads.test" || blocked[0].Count != 3 || blocked[0].Blocked != 3 { + t.Errorf("TopBlocked[0] = %+v, want ads.test count 3 blocked 3", blocked[0]) + } + if blocked[0].QTypeS != "A" { + t.Errorf("QTypeS = %q, want A", blocked[0].QTypeS) + } + queried := l.TopQueried(5) + if len(queried) != 2 { + t.Fatalf("TopQueried = %+v, want two domains", queried) + } + if queried[0].Name != "ads.test" || queried[0].Count != 3 { + t.Errorf("TopQueried[0] = %+v, want ads.test count 3 first", queried[0]) + } + if got := l.TopDomains(5, true); len(got) != 1 { + t.Errorf("TopDomains(blockedOnly) = %+v, want one entry", got) + } +} + +func BenchmarkLog(b *testing.B) { + l, err := New(Options{Path: "", LiveBuffer: 4096, MaxEntries: 1024, TopDomains: 512, TopTTL: time.Minute}) + if err != nil { + b.Fatal(err) + } + defer func() { _ = l.Close() }() + e := entryFor(1) + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + l.Log(e) + } +} + +func BenchmarkRecent(b *testing.B) { + l, err := New(Options{Path: "", LiveBuffer: 4096, MaxEntries: 1024, TopDomains: 512, TopTTL: time.Minute}) + if err != nil { + b.Fatal(err) + } + defer func() { _ = l.Close() }() + for i := 0; i < 4096; i++ { + l.Log(entryFor(i%2000 + 1)) + } + b.ReportAllocs() + b.ResetTimer() + for i := 0; i < b.N; i++ { + _ = l.Recent(200) + } +} + +// TestEntriesWithNoTimestampAreStamped guards the LiveEntry field the SSE +// consumers render: an entry with no TS must not reach them as year 1. +func TestEntriesWithNoTimestampAreStamped(t *testing.T) { + l := openLogger(t, tempPath(t), func(o *Options) { + o.Path = "" + o.Now = func() time.Time { return time.Date(2024, 5, 1, 12, 0, 0, 0, time.UTC) } + }) + l.Log(Entry{Name: "clock.test.", QType: dns.TypeA, Action: ActionAllow}) + got := l.Recent(1) + if len(got) != 1 { + t.Fatalf("Recent(1) = %d entries, want 1", len(got)) + } + if !strings.HasSuffix(got[0].Name, "clock.test.") { + t.Errorf("Name = %q", got[0].Name) + } + want := time.Date(2024, 5, 1, 12, 0, 0, 0, time.UTC) + if !got[0].TS.Equal(want) { + t.Errorf("TS = %v, want the injected clock %v", got[0].TS, want) + } +} diff --git a/internal/zonefile/zonefile.go b/internal/zonefile/zonefile.go index cc8d9e0..d236e9a 100644 --- a/internal/zonefile/zonefile.go +++ b/internal/zonefile/zonefile.go @@ -48,6 +48,9 @@ type Zone struct { SOA *dns.SOA // DefaultTTL is the $TTL directive value, else the SOA minimum. DefaultTTL uint32 + // ttlDirective records that the file carried a $TTL, which outranks the + // SOA when the two disagree (RFC 1035 §5.2). + ttlDirective bool } // Options tunes parsing. @@ -101,7 +104,7 @@ func Parse(r io.Reader, opt Options) (*Zone, error) { lastOwner = z.Origin continue } - assembled, err := z.assemble(logical, lastOwner) + assembled, hasTTL, err := z.assemble(logical, lastOwner) if err != nil { return nil, fmt.Errorf("%s:%d: %w", where(opt.File), startLine, err) } @@ -109,6 +112,17 @@ func Parse(r io.Reader, opt Options) (*Zone, error) { if err != nil { return nil, fmt.Errorf("%s:%d: %q: %w", where(opt.File), startLine, assembled, err) } + // RFC 1035 §5.2: a record with no TTL column takes the $TTL in force. + // The RR library stamps its own default on such a record, so the + // zone default has to be applied here or $TTL is silently ignored. + if !hasTTL && z.DefaultTTL > 0 { + rr.Header().Ttl = z.DefaultTTL + } + // RFC 1035 §5.1: a domain name in the RDATA section that is written + // without a trailing dot is relative to the zone origin, not to the + // root. Left to the library it becomes "host.", which points a CNAME + // at the root instead of at the zone. + qualifyRdata(rr, z.Origin) rec := Record{ Name: strings.ToLower(ensureDot(rr.Header().Name)), TTL: rr.Header().Ttl, @@ -124,10 +138,14 @@ func Parse(r io.Reader, opt Options) (*Zone, error) { } if soa, ok := rr.(*dns.SOA); ok { z.SOA = cloneSOA(soa) - if rec.TTL > 0 { - z.DefaultTTL = rec.TTL - } else if z.DefaultTTL == 0 { - z.DefaultTTL = soa.Minttl + // An explicit $TTL wins; the SOA only supplies the default for a + // file that never set one. + if !z.ttlDirective { + if rec.TTL > 0 { + z.DefaultTTL = rec.TTL + } else if z.DefaultTTL == 0 { + z.DefaultTTL = soa.Minttl + } } } z.Records = append(z.Records, rec) @@ -169,12 +187,15 @@ func deparen(s string) string { return strings.TrimSpace(s) } -// assemble fills in the omitted owner name, TTL and class so that the result -// is a complete RR line that dns.NewRR will accept. -func (z *Zone) assemble(line, lastOwner string) (string, error) { +// assemble fills in the omitted owner name and class so that the result is a +// complete RR line that dns.NewRR will accept. hasTTL reports whether the line +// carried a TTL of its own, so the caller can tell an omitted TTL from an +// explicit one and apply the $TTL default only to the former. +func (z *Zone) assemble(line, lastOwner string) (string, bool, error) { + var hasTTL bool fields := splitFields(line) if len(fields) == 0 { - return "", fmt.Errorf("empty record") + return "", false, fmt.Errorf("empty record") } var head string var rest []string @@ -187,7 +208,7 @@ func (z *Zone) assemble(line, lastOwner string) (string, error) { rest = fields[1:] } if head == "" { - return "", fmt.Errorf("record has no owner name and no previous owner to inherit") + return "", false, fmt.Errorf("record has no owner name and no previous owner to inherit") } head = relative(head, z.Origin) @@ -197,18 +218,19 @@ func (z *Zone) assemble(line, lastOwner string) (string, error) { switch { case isTTLToken(rest[0]): out = append(out, rest[0]) + hasTTL = true case isClassToken(rest[0]): out = append(out, strings.ToUpper(rest[0])) default: - return "", fmt.Errorf("unexpected token %q before the record type", rest[0]) + return "", false, fmt.Errorf("unexpected token %q before the record type", rest[0]) } rest = rest[1:] } if len(rest) == 0 { - return "", fmt.Errorf("record %q has no type", head) + return "", false, fmt.Errorf("record %q has no type", head) } out = append(out, rest...) - return strings.Join(out, " "), nil + return strings.Join(out, " "), hasTTL, nil } // directive handles $TTL, $ORIGIN and $INCLUDE. @@ -224,6 +246,7 @@ func (z *Zone) directive(line string, opt Options) error { return err } z.DefaultTTL = ttl + z.ttlDirective = true case "$ORIGIN": if len(fields) < 2 { return fmt.Errorf("$ORIGIN needs a name") @@ -267,6 +290,48 @@ func cloneSOA(in *dns.SOA) *dns.SOA { return &out } +// qualifyRdata appends the origin to the domain names in a record's RDATA +// that were written without a trailing dot. RFC 1035 §5.1 makes those names +// relative to the zone origin, so "www CNAME host" in example.com. means +// host.example.com. and not host. A name that already ends in a dot is +// absolute and left alone, which keeps the RPZ targets "." and "rpz-drop." +// and the null MX/SRV targets intact. +func qualifyRdata(rr dns.RR, origin string) { + switch v := rr.(type) { + case *dns.CNAME: + v.Target = relative(v.Target, origin) + case *dns.DNAME: + v.Target = relative(v.Target, origin) + case *dns.NS: + v.Ns = relative(v.Ns, origin) + case *dns.PTR: + v.Ptr = relative(v.Ptr, origin) + case *dns.MX: + v.Mx = relative(v.Mx, origin) + case *dns.SRV: + v.Target = relative(v.Target, origin) + case *dns.SOA: + v.Ns = relative(v.Ns, origin) + v.Mbox = relative(v.Mbox, origin) + case *dns.NAPTR: + v.Replacement = relative(v.Replacement, origin) + case *dns.RP: + v.Mbox = relative(v.Mbox, origin) + v.Txt = relative(v.Txt, origin) + case *dns.MINFO: + v.Rmail = relative(v.Rmail, origin) + v.Email = relative(v.Email, origin) + case *dns.MB: + v.Mb = relative(v.Mb, origin) + case *dns.MG: + v.Mg = relative(v.Mg, origin) + case *dns.MR: + v.Mr = relative(v.Mr, origin) + case *dns.KX: + v.Exchanger = relative(v.Exchanger, origin) + } +} + func isRRTYPE(s string) bool { return rrTypes[strings.ToUpper(s)] } func isTTLToken(s string) bool {