over-engineering audit: remove dead code, speculative config, duplicate queues

- delete embedded community resolver snapshot (6,760 lines) + EmbeddedServers +
  DNS_SERVERS=fresh; curated resolvers are the default. Simplify Warm (no
  sampling).
- one generic internal/sample.Queue[T] replaces three near-identical
  random-sample queues (probe, pipeline, discover).
- delete unreachable index.listOpenSearch (OpenSearch is never listable).
- delete classifyCloudFront (its match is always discarded, !ResourceSpecific).
- delete probe.Probe wrapper + Describe; tests use ProbeDetailed.
- delete store.Pool(), model.SourceDNSAlias, discover.QueueLen.
- drop dead config/no-op knobs: pipeline DirectWorkers, discover MaxPerDomain,
  discover S3/GCS/DO flags, BUCKET_SCAN_DO.
~-7,000 lines (mostly the resolver snapshot), 0 deps removed, tests green.

internal/pipeline/dnsqueue.go

@@ -1,71 +0,0 @@
-package pipeline
-
-import (
-	mathrand "math/rand/v2"
-	"sync"
-)
-
-// stringSample is a memory-bounded queue that keeps a random sample of recent
-// hostnames instead of a FIFO. A FIFO fills with the first N hostnames seen at
-// startup and drops everything after, so DNS enrichment only ever sees stale
-// hostnames and CNAME discovery stalls. Random replacement makes coverage
-// representative at bounded memory. Not on the certificate hot path.
-type stringSample struct {
-	mu     sync.Mutex
-	items  []string
-	cap    int
-	notify chan struct{}
-	stop   chan struct{}
-	once   sync.Once
-}
-
-func newStringSample(capacity int) *stringSample {
-	return &stringSample{
-		cap:    capacity,
-		notify: make(chan struct{}, 1),
-		stop:   make(chan struct{}),
-	}
-}
-
-func (q *stringSample) push(s string) {
-	q.mu.Lock()
-	if len(q.items) < q.cap {
-		q.items = append(q.items, s)
-	} else {
-		q.items[mathrand.IntN(len(q.items))] = s
-	}
-	q.mu.Unlock()
-	select {
-	case q.notify <- struct{}{}:
-	default:
-	}
-}
-
-func (q *stringSample) pop() (string, bool) {
-	q.mu.Lock()
-	n := len(q.items)
-	if n == 0 {
-		q.mu.Unlock()
-		return "", false
-	}
-	i := mathrand.IntN(n)
-	s := q.items[i]
-	q.items[i] = q.items[n-1]
-	q.items = q.items[:n-1]
-	q.mu.Unlock()
-	return s, true
-}
-
-func (q *stringSample) len() int {
-	q.mu.Lock()
-	defer q.mu.Unlock()
-	return len(q.items)
-}
-
-func (q *stringSample) close() {
-	q.once.Do(func() { close(q.stop) })
-	select {
-	case q.notify <- struct{}{}:
-	default:
-	}
-}

internal/pipeline/pipeline.go

@@ -15,6 +15,7 @@ import (
 	"tea.karlcom.de/karl/bucketeer/internal/dns"
 	"tea.karlcom.de/karl/bucketeer/internal/metrics"
 	"tea.karlcom.de/karl/bucketeer/internal/model"
+	"tea.karlcom.de/karl/bucketeer/internal/sample"
 )
 
 // Finding is a fully enriched candidate ready to be persisted.
@@ -32,7 +33,6 @@ type Config struct {
 	DirectQueueSize int
 	DNSQueueSize    int
 	DBQueueSize     int
-	DirectWorkers   int
 	DNSWorkers      int
 	BatchSize       int
 	FlushInterval   time.Duration
@@ -54,7 +54,7 @@ type Pipeline struct {
 	cfg Config
 
 	directQ chan model.Candidate
-	dnsQ    *stringSample
+	dnsQ    *sample.Queue[string]
 	dbQ     chan Finding
 
 	dedup    *dns.TTLCache[bool] // hostnames already queued for DNS
@@ -73,9 +73,6 @@ func New(cfg Config) *Pipeline {
 	if cfg.DBQueueSize <= 0 {
 		cfg.DBQueueSize = 8192
 	}
-	if cfg.DirectWorkers <= 0 {
-		cfg.DirectWorkers = 2
-	}
 	if cfg.DNSWorkers <= 0 {
 		cfg.DNSWorkers = 16
 	}
@@ -94,14 +91,14 @@ func New(cfg Config) *Pipeline {
 	p := &Pipeline{
 		cfg:     cfg,
 		directQ: make(chan model.Candidate, cfg.DirectQueueSize),
-		dnsQ:    newStringSample(cfg.DNSQueueSize),
+		dnsQ:    sample.New[string](cfg.DNSQueueSize),
 		dbQ:     make(chan Finding, cfg.DBQueueSize),
 		dedup:   dns.NewTTLCache[bool](cfg.DedupSize, 64),
 		seen:    dns.NewTTLCache[bool](cfg.DedupSize, 64),
 	}
 	if p.cfg.Metrics != nil {
 		p.cfg.Metrics.SetQueueDepth("direct", func() float64 { return float64(len(p.directQ)) })
-		p.cfg.Metrics.SetQueueDepth("dns", func() float64 { return float64(p.dnsQ.len()) })
+		p.cfg.Metrics.SetQueueDepth("dns", func() float64 { return float64(p.dnsQ.Len()) })
 		p.cfg.Metrics.SetQueueDepth("db", func() float64 { return float64(len(p.dbQ)) })
 	}
 	return p
@@ -109,7 +106,7 @@ func New(cfg Config) *Pipeline {
 
 // Start launches the worker pools. Close must be called to drain.
 func (p *Pipeline) Start(ctx context.Context) {
-	for i := 0; i < p.cfg.DirectWorkers; i++ {
+	for i := 0; i < 2; i++ {
 		p.producer.Add(1)
 		go p.directWorker(ctx)
 	}
@@ -123,14 +120,14 @@ func (p *Pipeline) Start(ctx context.Context) {
 
 // Depths reports the current length of the direct, DNS and DB queues.
 func (p *Pipeline) Depths() (direct, dns, db int) {
-	return len(p.directQ), p.dnsQ.len(), len(p.dbQ)
+	return len(p.directQ), p.dnsQ.Len(), len(p.dbQ)
 }
 
 // Close stops intake and drains all queues. It must only be called after the
 // CertSpotter monitor has stopped calling HandleIdentifiers.
 func (p *Pipeline) Close() {
 	close(p.directQ)
-	p.dnsQ.close()
+	p.dnsQ.Close()
 	p.producer.Wait()
 	close(p.dbQ)
 	p.workers.Wait()
@@ -179,7 +176,7 @@ func (p *Pipeline) HandleHost(host string) {
 		return
 	}
 	p.dedup.Set(host, true, p.cfg.DedupTTL)
-	p.dnsQ.push(host)
+	p.dnsQ.Push(host)
 	if p.cfg.Discover != nil {
 		p.cfg.Discover.Submit(host)
 	}
@@ -242,12 +239,12 @@ func (p *Pipeline) dnsWorker(ctx context.Context) {
 		select {
 		case <-ctx.Done():
 			return
-		case <-p.dnsQ.stop:
+		case <-p.dnsQ.Stop():
 			return
-		case <-p.dnsQ.notify:
+		case <-p.dnsQ.Notify():
 		}
 		for {
-			host, ok := p.dnsQ.pop()
+			host, ok := p.dnsQ.Pop()
 			if !ok {
 				break
 			}

internal/pipeline/pipeline_test.go

@@ -269,7 +269,7 @@ func TestDNSDedup(t *testing.T) {
 	p := New(Config{DNSQueueSize: 8, Metrics: m})
 	p.HandleHost("files.example.com")
 	p.HandleHost("files.example.com")
-	if got := p.dnsQ.len(); got != 1 {
+	if got := p.dnsQ.Len(); got != 1 {
 		t.Fatalf("dns queue len = %d, want 1", got)
 	}
 }