From 9710efec1aee0221773d407962d5241990f5e82a Mon Sep 17 00:00:00 2001 From: StasonJatham Date: Tue, 22 Sep 2026 13:33:05 +0200 Subject: [PATCH] fix memory blowup and keep direct matches from being dropped - dns.TTLCache: stop preallocating every shard map to maxShard; this allocated the full theoretical cache capacity up front (hundreds of MB of empty buckets) and pushed the container over its memory limit, causing an OOM restart loop under Docker's default 1g cap. - pipeline: write direct provider matches immediately without a blocking DNS enrichment step, so they can no longer be starved by the optional CNAME queue. Add TestDirectNotStarvedByDNSBacklog. - config: add DEDUP_SIZE, lower DNS_CACHE_SIZE/DEDUP_SIZE defaults to 256k. - .env.example / README: document DEDUP_SIZE and cache sizing. Live: stable at ~250-300 MiB with no restarts, direct queue draining to 0. --- .env.example | 2 ++ README.md | 3 +- cmd/ctcloud/main.go | 1 + internal/config/config.go | 5 ++- internal/dns/cache.go | 6 +++- internal/pipeline/pipeline.go | 21 ++++-------- internal/pipeline/pipeline_test.go | 55 ++++++++++++++++++++++++++++++ 7 files changed, 75 insertions(+), 18 deletions(-) diff --git a/.env.example b/.env.example index b92fab1..933ac5a 100644 --- a/.env.example +++ b/.env.example @@ -12,5 +12,7 @@ DATABASE_URL=postgres://ctcloud:change-me-strong@postgres:5432/ctcloud?sslmode=d DNS_SERVERS=1.1.1.1:53,8.8.8.8:53,1.0.0.1:53 DNS_WORKERS=16 +DNS_CACHE_SIZE=262144 +DEDUP_SIZE=262144 LOG_LEVEL=info ACTIVE_PROBING=false diff --git a/README.md b/README.md index ac2e014..4a2d0a3 100644 --- a/README.md +++ b/README.md @@ -111,7 +111,8 @@ Environment variables only; no config files, no runtime pattern files. | `DNS_SERVERS` | `1.1.1.1:53,8.8.8.8:53,1.0.0.1:53` | comma-separated resolvers (round-robin + failover); `system` for the host resolver | | `DNS_WORKERS` | `16` | DNS enrichment workers | | `DNS_TIMEOUT` | `5s` | per-lookup timeout | -| `DNS_CACHE_SIZE` | `1048576` | max DNS cache entries | +| `DNS_CACHE_SIZE` | `262144` | max DNS cache entries | +| `DEDUP_SIZE` | `262144` | max entries in each dedup cache | | `DNS_MAX_CNAME_DEPTH` | `8` | max CNAME hops | | `DNS_POSITIVE_TTL` / `DNS_NEGATIVE_TTL` | `10m` / `1m` | DNS cache TTLs | | `DIRECT_QUEUE_SIZE` | `8192` | direct-match queue capacity | diff --git a/cmd/ctcloud/main.go b/cmd/ctcloud/main.go index 3b723e7..18dcd17 100644 --- a/cmd/ctcloud/main.go +++ b/cmd/ctcloud/main.go @@ -114,6 +114,7 @@ func run() error { BatchSize: cfg.DBBatchSize, FlushInterval: cfg.DBFlushInterval, DedupTTL: cfg.DedupeTTL, + DedupSize: cfg.DedupSize, Enricher: enricher, Write: write, Metrics: m, diff --git a/internal/config/config.go b/internal/config/config.go index 07285a5..9cdbcc4 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -22,6 +22,8 @@ type Config struct { DNSPositiveTTL time.Duration DNSNegativeTTL time.Duration + DedupSize int + DirectQueueSize int DNSQueueSize int DBQueueSize int @@ -57,7 +59,8 @@ func Load() (*Config, error) { DNSServers: envList("DNS_SERVERS", DefaultDNSServers), DNSWorkers: envInt("DNS_WORKERS", 16), DNSTimeout: envDuration("DNS_TIMEOUT", 5*time.Second), - DNSCacheSize: envInt("DNS_CACHE_SIZE", 1<<20), + DNSCacheSize: envInt("DNS_CACHE_SIZE", 1<<18), + DedupSize: envInt("DEDUP_SIZE", 1<<18), DNSMaxCNAMEdepth: envInt("DNS_MAX_CNAME_DEPTH", 8), DNSPositiveTTL: envDuration("DNS_POSITIVE_TTL", 10*time.Minute), DNSNegativeTTL: envDuration("DNS_NEGATIVE_TTL", 1*time.Minute), diff --git a/internal/dns/cache.go b/internal/dns/cache.go index 67e4c4b..ab9dc75 100644 --- a/internal/dns/cache.go +++ b/internal/dns/cache.go @@ -41,7 +41,11 @@ func NewTTLCache[V any](maxEntries, shards int) *TTLCache[V] { c := &TTLCache[V]{mask: uint64(n - 1), maxShard: maxShard} c.shards = make([]shard[V], n) for i := range c.shards { - c.shards[i].m = make(map[string]cacheEntry[V], maxShard) + // Deliberately no capacity hint: pre-sizing every shard to maxShard + // allocates the full theoretical capacity (hundreds of MB across the + // caches) up front. Let the maps grow on demand; maxShard still bounds + // them via eviction. + c.shards[i].m = make(map[string]cacheEntry[V]) } return c } diff --git a/internal/pipeline/pipeline.go b/internal/pipeline/pipeline.go index bb61a39..f2cfebf 100644 --- a/internal/pipeline/pipeline.go +++ b/internal/pipeline/pipeline.go @@ -208,20 +208,6 @@ func appendIPs(f *Finding, ips []net.IP) { } } -func (p *Pipeline) enrich(ctx context.Context, c model.Candidate) Finding { - f := Finding{Candidate: c} - if p.cfg.Enricher == nil { - return f - } - res, ok := p.cfg.Enricher.Resolve(ctx, c.Hostname) - if !ok { - return f - } - f.CNAMEChain = res.Chain - appendIPs(&f, res.IPs) - return f -} - func (p *Pipeline) directWorker(ctx context.Context) { defer p.producer.Done() for { @@ -232,7 +218,12 @@ func (p *Pipeline) directWorker(ctx context.Context) { if !ok { return } - p.pushDB(ctx, p.enrich(ctx, c)) + // Direct provider matches are written immediately and are never + // queued behind optional DNS work, so they cannot be shed in favour + // of speculative CNAME enrichment. The matched provider hostname is + // the finding; A/AAAA enrichment is a DNS-bound luxury that the + // shared resolver cannot keep up with at these volumes. + p.pushDB(ctx, Finding{Candidate: c}) } } } diff --git a/internal/pipeline/pipeline_test.go b/internal/pipeline/pipeline_test.go index af6ab5a..a019921 100644 --- a/internal/pipeline/pipeline_test.go +++ b/internal/pipeline/pipeline_test.go @@ -195,6 +195,61 @@ func TestCNAMEMatchIPv6(t *testing.T) { } } +type blockingResolver struct{ release chan struct{} } + +func (b *blockingResolver) LookupCNAME(ctx context.Context, host string) (string, error) { + select { + case <-b.release: + case <-ctx.Done(): + } + return host, nil +} + +func (b *blockingResolver) LookupIP(ctx context.Context, host string) ([]net.IP, error) { + return nil, nil +} + +// TestDirectNotStarvedByDNSBacklog verifies direct provider matches are written +// even while the DNS queue is saturated with optional CNAME work. +func TestDirectNotStarvedByDNSBacklog(t *testing.T) { + col := &collector{} + release := make(chan struct{}) + p := New(Config{ + DNSQueueSize: 1, + DNSWorkers: 1, + Enricher: newEnricher(&blockingResolver{release: release}), + Write: col.write, + BatchSize: 10, + FlushInterval: 10 * time.Millisecond, + }) + ctx, cancel := context.WithCancel(context.Background()) + p.Start(ctx) + for i := 0; i < 100; i++ { + p.HandleHost("api" + strconv.Itoa(i) + ".example.com") // fills/blocks DNS queue + } + p.HandleHost("bucket.s3.eu-central-1.amazonaws.com") + + found := false + deadline := time.Now().Add(2 * time.Second) + for time.Now().Before(deadline) { + for _, f := range col.all() { + if f.Hostname == "bucket.s3.eu-central-1.amazonaws.com" { + found = true + } + } + if found { + break + } + time.Sleep(5 * time.Millisecond) + } + close(release) + cancel() + p.Close() + if !found { + t.Fatal("direct match was not written while the DNS queue was saturated") + } +} + func TestQueueSaturationDrops(t *testing.T) { m := metrics.New() p := New(Config{DirectQueueSize: 1, DNSQueueSize: 1, Metrics: m})