parallel backfill, persist scanned exposure, random-sample DNS queue

- scanner.run now persists exposure_state/detail/capabilities for scanned
  assets (previously only the index was written).
- Backfill runs a bounded worker pool (16) so it verifies reachable assets
  without saturating the DB pool; DB_MAX_CONNS raised for the live test.
- CNAME enrichment queue is now a bounded random sample instead of a FIFO.
  The FIFO filled with the first N hostnames seen at startup and dropped the
  rest, so CNAME discovery only ever saw stale hostnames; random replacement
  gives representative coverage at bounded memory (same fix as the prober).

Verified live:
- all four known-public resources scan + index (DO, AWS s3-<region>, GCS,
  Azure container path), 38-1000 objects each.
- backfill of every CT-discovered 'reachable' asset (725) and 'auth_required'
  (12) completed with 0 listable: those are MRAP/non-bucket 400s and
  list-denied buckets, confirming the checks are actually running.

cmd/ctcloud/main.go

@@ -135,7 +135,13 @@ func run() error {
 	// Scanner prober: a separate, never-started Prober used synchronously by the
 	// on-demand scan endpoint (Probe is stateless).
 	scanProber := probe.New(probe.Config{Workers: 1, Timeout: cfg.ProbeTimeout, Deep: cfg.ProbeDeep, Metrics: m})
-	scanner := &scan.Scanner{Prober: scanProber, Indexer: indexer, Enricher: enricher, Store: st}
+	// Backfill does one DB write per target, so keep its concurrency near the
+	// pool size to avoid starving the live pipeline and the UI.
+	backfillWorkers := cfg.ProbeWorkers
+	if backfillWorkers > 16 {
+		backfillWorkers = 16
+	}
+	scanner := &scan.Scanner{Prober: scanProber, Indexer: indexer, Enricher: enricher, Store: st, BackfillWorkers: backfillWorkers}
 
 	var prober *probe.Prober
 	if cfg.ActiveProbing {

internal/pipeline/dnsqueue.go

@@ -0,0 +1,71 @@
+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

@@ -50,7 +50,7 @@ type Pipeline struct {
 	cfg Config
 
 	directQ chan model.Candidate
-	dnsQ    chan string
+	dnsQ    *stringSample
 	dbQ     chan Finding
 
 	dedup    *dns.TTLCache[bool] // hostnames already queued for DNS
@@ -90,14 +90,14 @@ func New(cfg Config) *Pipeline {
 	p := &Pipeline{
 		cfg:     cfg,
 		directQ: make(chan model.Candidate, cfg.DirectQueueSize),
-		dnsQ:    make(chan string, cfg.DNSQueueSize),
+		dnsQ:    newStringSample(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(len(p.dnsQ)) })
+		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
@@ -119,14 +119,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), len(p.dnsQ), 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)
-	close(p.dnsQ)
+	p.dnsQ.close()
 	p.producer.Wait()
 	close(p.dbQ)
 	p.workers.Wait()
@@ -175,11 +175,7 @@ func (p *Pipeline) HandleHost(host string) {
 		return
 	}
 	p.dedup.Set(host, true, p.cfg.DedupTTL)
-	select {
-	case p.dnsQ <- host:
-	default:
-		p.drop()
-	}
+	p.dnsQ.push(host)
 }
 
 func (p *Pipeline) drop() {
@@ -239,12 +235,17 @@ func (p *Pipeline) dnsWorker(ctx context.Context) {
 		select {
 		case <-ctx.Done():
 			return
-		case host, ok := <-p.dnsQ:
+		case <-p.dnsQ.stop:
+			return
+		case <-p.dnsQ.notify:
+		}
+		for {
+			host, ok := p.dnsQ.pop()
 			if !ok {
-				return
+				break
 			}
 			if len(p.dbQ) >= cap(p.dbQ)/2 {
-				// Load shedding: optional CNAME enrichment is dropped before
+				// Load shedding: optional CNAME enrichment is skipped before
 				// direct matches when the database writer is backed up.
 				p.drop()
 				continue

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 := len(p.dnsQ); got != 1 {
+	if got := p.dnsQ.len(); got != 1 {
 		t.Fatalf("dns queue len = %d, want 1", got)
 	}
 }

internal/scan/scanner.go

@@ -8,6 +8,8 @@ import (
 	"context"
 	"net"
 	"strings"
+	"sync"
+	"sync/atomic"
 
 	"tea.karlcom.de/karl/bucketeer/internal/classify"
 	"tea.karlcom.de/karl/bucketeer/internal/dns"
@@ -37,10 +39,11 @@ type Enricher interface {
 }
 
 type Scanner struct {
-	Prober   Prober
-	Indexer  Lister
-	Enricher Enricher
-	Store    Store
+	Prober          Prober
+	Indexer         Lister
+	Enricher        Enricher
+	Store           Store
+	BackfillWorkers int
 }
 
 // Result is the outcome of scanning one resolved provider resource.
@@ -107,47 +110,84 @@ type BackfillReport struct {
 
 // Backfill re-probes assets already in the database (optionally filtered by
 // exposure_state) with the current checks and indexes any that are readable.
-// This recovers findings discovered before a check was improved.
+// This is how already-discovered "reachable" endpoints actually get verified.
+// It runs a bounded worker pool so it stays interactive.
 func (s *Scanner) Backfill(ctx context.Context, exposure string, limit int) (BackfillReport, error) {
 	targets, err := s.Store.ListProbeTargets(ctx, exposure, limit)
 	if err != nil {
 		return BackfillReport{}, err
 	}
-	var rep BackfillReport
-	for _, t := range targets {
-		if err := ctx.Err(); err != nil {
-			return rep, err
-		}
-		c := model.Candidate{
-			Hostname: t.Hostname, Original: t.Hostname, Resource: t.Resource, Region: t.Region,
-			Provider: model.ParseProvider(t.Provider), Service: model.ParseService(t.Service),
-			Source: model.SourceManual,
-		}
-		pr := s.Prober.ProbeDetailed(ctx, c)
-		rep.Probed++
-		if pr.Exposure == model.ExposurePublicRead {
-			rep.Readable++
-		}
-		if pr.Listable {
-			rep.Listable++
-		}
-		if err := s.Store.UpdateExposure(ctx, t.Provider, t.Service, t.Hostname,
-			pr.Exposure, pr.Detail, pr.Listable, pr.ObjectReadable); err != nil {
-			rep.Errors++
-			continue
-		}
-		if pr.Listable && s.Indexer != nil {
-			entries, err := s.Indexer.List(ctx, c)
-			if err == nil && len(entries) > 0 {
-				if n, werr := s.Store.ReplaceEntries(ctx, t.Provider, t.Service, t.Hostname, entries); werr == nil {
-					rep.Indexed += n
-				} else {
-					rep.Errors++
+	workers := s.BackfillWorkers
+	if workers <= 0 {
+		workers = 32
+	}
+	var (
+		probed, readable, listable, indexed, errs atomic.Int64
+		wg                                        sync.WaitGroup
+		jobs                                      = make(chan model.ProbeTarget)
+	)
+	for i := 0; i < workers; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for t := range jobs {
+				if ctx.Err() != nil {
+					return
 				}
+				s.probeTarget(ctx, t, &probed, &readable, &listable, &indexed, &errs)
+			}
+		}()
+	}
+	for _, t := range targets {
+		select {
+		case jobs <- t:
+		case <-ctx.Done():
+			close(jobs)
+			wg.Wait()
+			return s.report(&probed, &readable, &listable, &indexed, &errs), ctx.Err()
+		}
+	}
+	close(jobs)
+	wg.Wait()
+	return s.report(&probed, &readable, &listable, &indexed, &errs), nil
+}
+
+func (s *Scanner) probeTarget(ctx context.Context, t model.ProbeTarget, probed, readable, listable, indexed, errs *atomic.Int64) {
+	c := model.Candidate{
+		Hostname: t.Hostname, Original: t.Hostname, Resource: t.Resource, Region: t.Region,
+		Provider: model.ParseProvider(t.Provider), Service: model.ParseService(t.Service),
+		Source: model.SourceManual,
+	}
+	pr := s.Prober.ProbeDetailed(ctx, c)
+	probed.Add(1)
+	if pr.Exposure == model.ExposurePublicRead {
+		readable.Add(1)
+	}
+	if pr.Listable {
+		listable.Add(1)
+	}
+	if err := s.Store.UpdateExposure(ctx, t.Provider, t.Service, t.Hostname,
+		pr.Exposure, pr.Detail, pr.Listable, pr.ObjectReadable); err != nil {
+		errs.Add(1)
+		return
+	}
+	if pr.Listable && s.Indexer != nil {
+		entries, err := s.Indexer.List(ctx, c)
+		if err == nil && len(entries) > 0 {
+			if n, werr := s.Store.ReplaceEntries(ctx, t.Provider, t.Service, t.Hostname, entries); werr == nil {
+				indexed.Add(int64(n))
+			} else {
+				errs.Add(1)
 			}
 		}
 	}
-	return rep, nil
+}
+
+func (s *Scanner) report(probed, readable, listable, indexed, errs *atomic.Int64) BackfillReport {
+	return BackfillReport{
+		Probed: int(probed.Load()), Readable: int(readable.Load()),
+		Listable: int(listable.Load()), Indexed: int(indexed.Load()), Errors: int(errs.Load()),
+	}
 }
 
 func (s *Scanner) run(ctx context.Context, c model.Candidate, chain []string) Result {
@@ -161,6 +201,10 @@ func (s *Scanner) run(ctx context.Context, c model.Candidate, chain []string) Re
 	pr := s.Prober.ProbeDetailed(ctx, c)
 	r.Exposure, r.Detail = pr.Exposure.String(), pr.Detail
 	r.Listable, r.ObjectReadable = pr.Listable, pr.ObjectReadable
+	if err := s.Store.UpdateExposure(ctx, c.Provider.String(), c.Service.String(), c.Hostname,
+		pr.Exposure, pr.Detail, pr.Listable, pr.ObjectReadable); err != nil {
+		r.Error = "store: " + err.Error()
+	}
 	if pr.Listable && s.Indexer != nil {
 		entries, err := s.Indexer.List(ctx, c)
 		if err == nil && len(entries) > 0 {

internal/webui/webui.go

@@ -8,9 +8,12 @@ import (
 	"embed"
 	"encoding/json"
 	"io/fs"
+	"log/slog"
 	"net/http"
 	"strconv"
 	"strings"
+	"sync/atomic"
+	"time"
 
 	"tea.karlcom.de/karl/bucketeer/internal/scan"
 	"tea.karlcom.de/karl/bucketeer/internal/store"
@@ -111,19 +114,33 @@ func New(s Store, cfg Config) http.Handler {
 			}
 			writeJSON(w, cfg.Scanner.Scan(r.Context(), target))
 		})
+		var backfillRunning atomic.Bool
 		mux.HandleFunc("/api/reprobe", func(w http.ResponseWriter, r *http.Request) {
 			if r.Method != http.MethodPost {
 				http.Error(w, "POST only", http.StatusMethodNotAllowed)
 				return
 			}
-			q := r.URL.Query()
-			limit, _ := strconv.Atoi(q.Get("limit"))
-			rep, err := cfg.Scanner.Backfill(r.Context(), strings.TrimSpace(q.Get("exposure")), limit)
-			if err != nil {
-				http.Error(w, err.Error(), http.StatusInternalServerError)
+			if !backfillRunning.CompareAndSwap(false, true) {
+				http.Error(w, "backfill already running", http.StatusConflict)
 				return
 			}
-			writeJSON(w, rep)
+			q := r.URL.Query()
+			limit, _ := strconv.Atoi(q.Get("limit"))
+			exposure := strings.TrimSpace(q.Get("exposure"))
+			go func() {
+				defer backfillRunning.Store(false)
+				start := time.Now()
+				rep, err := cfg.Scanner.Backfill(context.Background(), exposure, limit)
+				if err != nil {
+					slog.Warn("backfill failed", "event", "backfill", "error", err)
+					return
+				}
+				slog.Info("backfill complete", "event", "backfill", "exposure", exposure,
+					"probed", rep.Probed, "readable", rep.Readable, "listable", rep.Listable,
+					"indexed", rep.Indexed, "errors", rep.Errors, "took", time.Since(start).String())
+			}()
+			w.WriteHeader(http.StatusAccepted)
+			writeJSON(w, map[string]any{"started": true, "exposure": exposure, "limit": limit})
 		})
 	}
 	mux.HandleFunc("/api/stats", func(w http.ResponseWriter, r *http.Request) {