fix: resolve race conditions, shutdown leaks, and critical bugs across pipeline

Fix data races:
- Add sync.RWMutex to dedup hash table in capture, protecting concurrent
  read/write during packet processing and ticker cleanup
- Add domainFilterMu sync.RWMutex to domain skip/allow filtering, ensuring
  thread-safe hot-reload of domain lists
- Add splunkConnMu sync.RWMutex to protect global splunk connection map
- Convert all captureConfig methods from value to pointer receivers to
  prevent copying the embedded mutex

Fix critical bugs:
- packet.go: use packet.ip instead of stale local ip4 in defrag return
  path, which caused incorrect protocol detection and flow extraction
- util.go: MaskSize6 validation checked MaskSize4 for lower bound
- influx.go: metric counter used wrong name "stdoutSkipped"
- util.go: remove redundant nested if-err-not-nil in config parsing

Fix graceful shutdown:
- Add defer close(closeChannel) to all 12 output backends to prevent
  Close() from deadlocking on context cancellation
- Add case <-ctx.Done() to output worker loops (sentinel, splunk, zinc,
  victorialogs) that were missing it
- Add sync.WaitGroup for worker lifecycle tracking in influx, parquet,
  postgres, stdout, victorialogs outputs

Replace log.Fatal with recoverable errors:
- elastic, file, postgres, syslog, influx, functions.go: return errors
  instead of fataling, allowing the dispatch loop to gracefully remove
  failed outputs
- LoadDomainsCsv now returns error as 4th value; callers log and skip
  update on failure instead of crashing

Other fixes:
- outputs.go: non-blocking fan-out dispatch with dropped packet metrics
- packet.go: log and count silent decode errors via decodingErrors metric
- packet.go: single errgroup for all input handler workers
- livecap_windows.go: snaplen 1600 -> 65535
- parquet.go: iterate all DNS questions instead of only first
- functions.go: store suffix domains pre-reversed for efficient lookup,
  add 30s HTTP timeout for remote domain lists
- splunk.go: remove deprecated rand.Seed
- main.go: remove no-op ctx.Done(), bound spin-wait with ctx.Err()

Add tests for dedup concurrency, defrag path correctness, transport
processing (UDP/TCP/non-DNS), and decode error handling.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: mosajjal

cmd/dnsmonster/main.go

@@ -50,11 +50,9 @@ func handleInterrupt(ctx context.Context) {
 	go func() {
 		<-c
 		log.Infof("SIGINT Received. Stopping capture...")
-		go util.GlobalCancel()
-		go ctx.Done()
+		util.GlobalCancel()
 		<-time.After(4 * time.Second)
 		log.Fatal("emergency exit")
-		os.Exit(1)
 	}()
 }
 
@@ -92,14 +90,12 @@ func main() {
 	g.Go(func() error { capture.GlobalCaptureConfig.CheckFlagsAndStart(ctx); return nil })
 	// Set up output dispatch
 	var c chan util.DNSResult
-	for {
+	for ctx.Err() == nil {
 		c = capture.GlobalCaptureConfig.GetResultChannel()
-		if c == nil {
-			time.Sleep(10 * time.Millisecond)
-			continue
-		} else {
+		if c != nil {
 			break
 		}
+		time.Sleep(10 * time.Millisecond)
 	}
 
 	g.Go(func() error { return setupOutputs(ctx, &c) })

cmd/dnsmonster/outputs.go

@@ -17,12 +17,15 @@ package main
 
 import (
 	"context"
+	"fmt"
 	"time"
 
-	_ "github.com/mosajjal/dnsmonster/internal/output" // this will automatically set up all the outputs
-	"github.com/mosajjal/dnsmonster/internal/util"
+	metrics "github.com/rcrowley/go-metrics"
 	log "github.com/sirupsen/logrus"
 	"golang.org/x/sync/errgroup"
+
+	_ "github.com/mosajjal/dnsmonster/internal/output" // this will automatically set up all the outputs
+	"github.com/mosajjal/dnsmonster/internal/util"
 )
 
 // a helper function to remove one of the outputs from the globaldispatch list
@@ -48,9 +51,9 @@ func setupOutputs(ctx context.Context, resultChannel *chan util.DNSResult) error
 		}
 	}
 
-	// check to see if at least one output is specified, otherwise we should panic exit
+	// check to see if at least one output is specified
 	if len(util.GlobalDispatchList) == 0 {
-		log.Fatal("No output specified. Please specify at least one output")
+		return fmt.Errorf("no output specified, please specify at least one output")
 	}
 	// todo: currently, there's no check to see if allowdomains and skipdomains are provided if the output type demands it.
 
@@ -72,21 +75,22 @@ func setupOutputs(ctx context.Context, resultChannel *chan util.DNSResult) error
 		log.Infof("allowDomains refresh interval is %s", util.GeneralFlags.AllowDomainsRefreshInterval)
 	}
 	g, gCtx := errgroup.WithContext(ctx)
+	dispatchedPackets := metrics.GetOrRegisterCounter("dispatchedPackets", metrics.DefaultRegistry)
+	droppedPackets := metrics.GetOrRegisterCounter("droppedPackets", metrics.DefaultRegistry)
+
 	g.Go(func() error {
 		// blocking loop
 		for {
 			select {
 			case data := <-*resultChannel:
 				for _, o := range util.GlobalDispatchList {
-					// Non-blocking send with timeout to prevent blocking on full channels
+					// Non-blocking send to prevent blocking on full channels
 					select {
 					case o.OutputChannel() <- data:
-						// Successfully sent
-					case <-time.After(100 * time.Millisecond):
-						// Channel is full or blocked, log and continue
-						log.Warnf("Output channel blocked, dropping packet")
-					case <-gCtx.Done():
-						return nil
+						dispatchedPackets.Inc(1)
+					default:
+						// Channel is full, drop packet
+						droppedPackets.Inc(1)
 					}
 				}
 
@@ -99,7 +103,7 @@ func setupOutputs(ctx context.Context, resultChannel *chan util.DNSResult) error
 			}
 		}
 	})
-	return nil
+	return g.Wait()
 }
 
 // vim: foldmethod=marker

internal/capture/afpacket_linux.go

@@ -92,7 +92,7 @@ func afpacketComputeSize(targetSizeMb uint, snaplen uint, pageSize uint) (
 // 	return err
 // }
 
-func (config captureConfig) initializeLiveAFpacket(devName, filter string) *afpacketHandle {
+func (config *captureConfig) initializeLiveAFpacket(devName, filter string) *afpacketHandle {
 	// Open device
 	// var tPacket *afpacket.TPacket
 	var err error

internal/capture/afpacket_nonlinux.go

@@ -59,7 +59,7 @@ func (afhandle *afpacketHandle) Stat() (uint, uint, error) {
 	return 0, 0, fmt.Errorf("Dnsmonster has been compiled without afpacket support for this platform")
 }
 
-func (config captureConfig) initializeLiveAFpacket(devName, filter string) *afpacketHandle {
+func (config *captureConfig) initializeLiveAFpacket(devName, filter string) *afpacketHandle {
 	return nil
 }
 

internal/capture/capture.go

@@ -69,6 +69,7 @@ type captureConfig struct {
 	ratioA                     int
 	ratioB                     int
 	dedupHashTable             map[uint64]bool
+	dedupMu                    sync.RWMutex
 }
 
 // GlobalCaptureConfig is accessible globally
@@ -84,11 +85,11 @@ func init() {
 
 }
 
-func (config captureConfig) GetResultChannel() chan util.DNSResult {
+func (config *captureConfig) GetResultChannel() chan util.DNSResult {
 	return config.resultChannel
 }
 
-func (config captureConfig) cleanExit(ctx context.Context) {
+func (config *captureConfig) cleanExit(ctx context.Context) {
 	ctx.Done()
 	log.Infof("Stopping capture...")
 }
@@ -139,7 +140,9 @@ func (config *captureConfig) CheckFlagsAndStart(ctx context.Context) {
 				select {
 				case <-time.NewTicker(config.DedupCleanupInterval).C:
 					log.Infof("cleaning up dedup hash table")
+					config.dedupMu.Lock()
 					config.dedupHashTable = make(map[uint64]bool)
+					config.dedupMu.Unlock()
 				case <-gCtx.Done():
 					return nil
 				}

internal/capture/capture_test.go

@@ -18,7 +18,9 @@ package capture
 import (
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
+	"time"
 )
 
 func TestSampleRatioParsing(t *testing.T) {
@@ -236,6 +238,109 @@ func TestPortValidation(t *testing.T) {
 	}
 }
 
+// TestDedupHashTableConcurrency verifies that concurrent reads and writes
+// to the dedup hash table are safe when protected by the mutex.
+// Must pass with -race.
+func TestDedupHashTableConcurrency(t *testing.T) {
+	config := &captureConfig{
+		dedupHashTable: make(map[uint64]bool),
+	}
+
+	const goroutines = 10
+	const iterations = 1000
+
+	var wg sync.WaitGroup
+
+	// Launch writer goroutines
+	for i := 0; i < goroutines; i++ {
+		wg.Add(1)
+		go func(id int) {
+			defer wg.Done()
+			for j := 0; j < iterations; j++ {
+				hash := FNV1A([]byte{byte(id), byte(j), byte(j >> 8)})
+				config.dedupMu.RLock()
+				_, exists := config.dedupHashTable[hash]
+				config.dedupMu.RUnlock()
+				if !exists {
+					config.dedupMu.Lock()
+					config.dedupHashTable[hash] = true
+					config.dedupMu.Unlock()
+				}
+			}
+		}(i)
+	}
+
+	// Launch a goroutine that clears the table (simulating the ticker)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for j := 0; j < 10; j++ {
+			config.dedupMu.Lock()
+			config.dedupHashTable = make(map[uint64]bool)
+			config.dedupMu.Unlock()
+			time.Sleep(time.Millisecond)
+		}
+	}()
+
+	wg.Wait()
+
+	// If we got here without a race detector complaint, the test passes
+	config.dedupMu.RLock()
+	size := len(config.dedupHashTable)
+	config.dedupMu.RUnlock()
+	t.Logf("Final dedup table size: %d", size)
+}
+
+// TestDedupHashTableClearing verifies the clearing mechanism works:
+// entries are removed when the table is cleared.
+func TestDedupHashTableClearing(t *testing.T) {
+	config := &captureConfig{
+		dedupHashTable: make(map[uint64]bool),
+	}
+
+	// Add some entries
+	for i := uint64(0); i < 100; i++ {
+		config.dedupMu.Lock()
+		config.dedupHashTable[i] = true
+		config.dedupMu.Unlock()
+	}
+
+	config.dedupMu.RLock()
+	if len(config.dedupHashTable) != 100 {
+		config.dedupMu.RUnlock()
+		t.Fatalf("Expected 100 entries, got %d", len(config.dedupHashTable))
+	}
+	config.dedupMu.RUnlock()
+
+	// Clear the table (simulating what the ticker does)
+	config.dedupMu.Lock()
+	config.dedupHashTable = make(map[uint64]bool)
+	config.dedupMu.Unlock()
+
+	config.dedupMu.RLock()
+	if len(config.dedupHashTable) != 0 {
+		config.dedupMu.RUnlock()
+		t.Fatalf("Expected 0 entries after clearing, got %d", len(config.dedupHashTable))
+	}
+	config.dedupMu.RUnlock()
+
+	// Verify we can add entries again after clearing
+	config.dedupMu.Lock()
+	config.dedupHashTable[42] = true
+	config.dedupMu.Unlock()
+
+	config.dedupMu.RLock()
+	if len(config.dedupHashTable) != 1 {
+		config.dedupMu.RUnlock()
+		t.Fatalf("Expected 1 entry after re-adding, got %d", len(config.dedupHashTable))
+	}
+	_, exists := config.dedupHashTable[42]
+	config.dedupMu.RUnlock()
+	if !exists {
+		t.Fatal("Expected key 42 to exist")
+	}
+}
+
 // Benchmark dedup hash lookup
 func BenchmarkDedupLookup(b *testing.B) {
 	dedupTable := make(map[uint64]bool)

internal/capture/dnstap.go

@@ -119,7 +119,7 @@ func dnsTapMsgToDNSResult(msg []byte) (*util.DNSResult, error) {
 	return &myDNSResult, nil
 }
 
-func (config captureConfig) StartDNSTap(ctx context.Context) error {
+func (config *captureConfig) StartDNSTap(ctx context.Context) error {
 	log.Info("Starting DNStap capture")
 
 	packetsCaptured := metrics.GetOrRegisterGauge("packetsCaptured", metrics.DefaultRegistry)

internal/capture/livecap_windows.go

@@ -29,7 +29,7 @@ type livePcapHandle struct {
 }
 
 func initializeLivePcap(devName, filter string) (*livePcapHandle, error) {
-	handle, err := pcap.OpenLive(devName, 1600, true, pcap.BlockForever)
+	handle, err := pcap.OpenLive(devName, 65535, true, pcap.BlockForever)
 	if err != nil {
 		return nil, fmt.Errorf("failed to open live capture on %s: %w", devName, err)
 	}

internal/capture/nondnstap.go

@@ -25,7 +25,7 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
-func (config captureConfig) StartNonDNSTap(ctx context.Context) error {
+func (config *captureConfig) StartNonDNSTap(ctx context.Context) error {
 	packetsCaptured := metrics.GetOrRegisterGauge("packetsCaptured", metrics.DefaultRegistry)
 	packetsDropped := metrics.GetOrRegisterGauge("packetsDropped", metrics.DefaultRegistry)
 	packetsDuplicate := metrics.GetOrRegisterCounter("packetsDuplicate", metrics.DefaultRegistry)
@@ -117,9 +117,13 @@ func (config captureConfig) StartNonDNSTap(ctx context.Context) error {
 		skipForDudup := false
 		if config.Dedup {
 			hash := FNV1A(data)
+			config.dedupMu.RLock()
 			_, ok := config.dedupHashTable[hash] // check for existence
+			config.dedupMu.RUnlock()
 			if !ok {
+				config.dedupMu.Lock()
 				config.dedupHashTable[hash] = true
+				config.dedupMu.Unlock()
 			} else {
 				skipForDudup = true
 				packetsDuplicate.Inc(1)