@marcel91200 / caddy Mirror · read-only / commits / bbea66f2
Fork 0

Fast and extensible multi-platform HTTP/1-2-3 web server with automatic HTTPS

Mirrored from https://github.com/Slazee/caddy.git

Synced Nov 11, 01:00 UTC

reverseproxy: reduce lock contention in dynamic SRV and A upstreams (#7949)

* reverseproxy: reduce lock contention in dynamic SRV and A upstreams (#7946) The shared SRV and A/AAAA caches held a global write lock across the DNS lookup, so one slow or timing-out record blocked every unrelated upstream source. Lookups now run without the cache lock held, deduplicated per key with singleflight. A per-key generation (stored on the cache entry, so it is bounded by the cache cap) and a cache-wide epoch keep a lookup that was in flight during ResetSRV/ResetAllSRV from writing its result back, and are part of the singleflight key so callers after a reset do not join a pre-reset lookup. The shared lookup is detached from the request that started it and bounded by its own 30s timeout; each caller still returns as soon as its own context is done. Distinct lookups across both caches are capped by a shared gate (64 running, 256 queued); beyond that they are shed, serving a stale SRV entry if a grace period is set and one is cached. SRV and A now share one cache implementation. AUpstreams also implements CachingUpstreamSource (via new ResetA/ResetAllA), so retries purge its cache entry like SRVUpstreams already did. Rebased onto the move of this code into modules/internal/network (#7790, #8252). Fixes #7946 Co-authored-by: Francis Lavoie <lavofr@gmail.com> Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * network: backdate cache entries in TestShedLookups instead of a 1ns refresh On Windows two close time.Now() calls can report no elapsed time, so an entry seeded with a 1ns refresh could still be fresh and skip the shed path entirely. * network: move the shared cache and lookup gate into cache.go dynamicupstreams.go keeps the SRV/A API; the generic cache, its singleflight/generation handling and the lookup gate are not specific to either record type. No functional change. * network: preserve cache reset and grace semantics --------- Co-authored-by: Francis Lavoie <lavofr@gmail.com> Co-authored-by: steadytao <mail@steadytao.com>

Faiyaz Rahman committed Nov 3, 2026 at 14:37 UTC bbea66f2ecec7907d09243801362546b84003706
5 files changed +815 -175
modules/caddyhttp/reverseproxy/upstreams.go
+21 -5
@@ -290,12 +290,20 @@ func (au *AUpstreams) Provision(ctx caddy.Context) error {
290 return nil
291 }
292
293 -func (au AUpstreams) GetUpstreams(r *http.Request) ([]*Upstream, error) {
294 - repl := r.Context().Value(caddy.ReplacerCtxKey).(*caddy.Replacer)
293 +// ResetCache clears the cached lookup for the given request, or the entire
294 +// A/AAAA cache if r is nil.
295 +func (au *AUpstreams) ResetCache(r *http.Request) error {
296 + if r == nil {
297 + network.ResetAllA()
298 + return nil
299 + }
300 + ipVersion, name, port := au.expandedAddr(r)
301 + network.ResetA(au.resolverKey, ipVersion, name, port)
302 + return nil
303 +}
304
296 - ipVersion := resolveIpVersion(au.Versions)
297 - name := repl.ReplaceAll(au.Name, "")
298 - port := repl.ReplaceAll(au.Port, "")
305 +func (au AUpstreams) GetUpstreams(r *http.Request) ([]*Upstream, error) {
306 + ipVersion, name, port := au.expandedAddr(r)
307
308 targets, err := network.A(r.Context(), au.resolver.LookupIP,
309 au.resolverKey, ipVersion, name, port, time.Duration(au.Refresh), au.logger)
@@ -312,6 +320,13 @@ func (au AUpstreams) GetUpstreams(r *http.Request) ([]*Upstream, error) {
320
321 func (au AUpstreams) String() string { return net.JoinHostPort(au.Name, au.Port) }
322
323 +// expandedAddr returns the IP version to resolve and the placeholder-expanded
324 +// name and port, which together identify the request's cache entry.
325 +func (au AUpstreams) expandedAddr(r *http.Request) (ipVersion, name, port string) {
326 + repl := r.Context().Value(caddy.ReplacerCtxKey).(*caddy.Replacer)
327 + return resolveIpVersion(au.Versions), repl.ReplaceAll(au.Name, ""), repl.ReplaceAll(au.Port, "")
328 +}
329 +
330 // MultiUpstreams is a single dynamic upstream source that
331 // aggregates the results of multiple dynamic upstream sources.
332 // All configured sources will be queried in order, with their
@@ -452,4 +467,5 @@ var (
467 _ CachingUpstreamSource = (*SRVUpstreams)(nil)
468 _ caddy.Provisioner = (*AUpstreams)(nil)
469 _ UpstreamSource = (*AUpstreams)(nil)
470 + _ CachingUpstreamSource = (*AUpstreams)(nil)
471 )
modules/internal/network/cache.go new
+258
@@ -0,0 +1,258 @@
1 +// Copyright 2015 Matthew Holt and The Caddy Authors
2 +//
3 +// Licensed under the Apache License, Version 2.0 (the "License");
4 +// you may not use this file except in compliance with the License.
5 +// You may obtain a copy of the License at
6 +//
7 +// http://www.apache.org/licenses/LICENSE-2.0
8 +//
9 +// Unless required by applicable law or agreed to in writing, software
10 +// distributed under the License is distributed on an "AS IS" BASIS,
11 +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 +// See the License for the specific language governing permissions and
13 +// limitations under the License.
14 +
15 +package network
16 +
17 +import (
18 + "context"
19 + "errors"
20 + "strconv"
21 + "sync"
22 + "sync/atomic"
23 + "time"
24 +
25 + "go.uber.org/zap"
26 + "go.uber.org/zap/zapcore"
27 + "golang.org/x/sync/singleflight"
28 +)
29 +
30 +const (
31 + // maxCacheEntries is the maximum number of entries kept in each cache.
32 + maxCacheEntries = 100
33 +
34 + // lookupTimeout bounds a shared lookup, which cannot use any one caller's
35 + // deadline.
36 + lookupTimeout = 30 * time.Second
37 +
38 + // maxLookups is the maximum number of distinct lookups allowed to run at
39 + // once, across both caches. Callers that join a lookup already in flight
40 + // do not count against it.
41 + maxLookups = 64
42 +
43 + // maxPendingLookups is the maximum number of lookups allowed to wait for a
44 + // slot. Beyond that they are shed rather than parked, so a stream of
45 + // distinct keys cannot pile up goroutines for up to lookupTimeout each.
46 + maxPendingLookups = 256
47 +)
48 +
49 +// errTooManyLookups is returned when the lookup gate is saturated.
50 +var errTooManyLookups = errors.New("too many dynamic upstream lookups in flight")
51 +
52 +// lookups is shared by every cache, so total resolver work stays bounded no
53 +// matter how many distinct keys come in.
54 +var lookups = newLookupGate(maxLookups, maxPendingLookups)
55 +
56 +// cache holds the results of one kind of lookup, and deduplicates concurrent
57 +// lookups for the same key.
58 +type cache struct {
59 + kind string // "SRV" or "A", for log messages
60 +
61 + mu sync.RWMutex
62 + entries map[string]cacheEntry
63 + epoch uint64 // bumped by every reset; guarded by mu
64 +
65 + group singleflight.Group
66 +}
67 +
68 +type cacheEntry struct {
69 + refresh time.Duration
70 + freshness time.Time
71 + targets []Target
72 +
73 + // gen counts targeted resets of this key. A lookup writes its result back
74 + // only if gen (and the cache's epoch) are unchanged since it started, and
75 + // both are part of the singleflight key, so a caller that arrives after a
76 + // reset starts its own lookup instead of joining one from before it. It
77 + // lives on the entry, rather than in a separate map, so it is bounded by
78 + // maxCacheEntries like everything else.
79 + gen uint64
80 +}
81 +
82 +func newCache(kind string) *cache {
83 + return &cache{kind: kind, entries: make(map[string]cacheEntry)}
84 +}
85 +
86 +func (e cacheEntry) isFresh() bool {
87 + return !e.freshness.IsZero() && time.Since(e.freshness) < e.refresh
88 +}
89 +
90 +// get returns the cached targets for key if they are fresh; otherwise it runs
91 +// lookup, sharing it with concurrent callers for the same key. See [SRV] for
92 +// the meaning of grace.
93 +func (c *cache) get(ctx context.Context, key string, refresh, grace time.Duration, logger *zap.Logger, lookup func(context.Context) ([]Target, error)) ([]Target, error) {
94 + c.mu.RLock()
95 + cached := c.entries[key]
96 + epoch := c.epoch
97 + c.mu.RUnlock()
98 + if cached.isFresh() {
99 + return cached.targets, nil
100 + }
101 + gen := cached.gen
102 +
103 + flightKey := strconv.FormatUint(epoch, 10) + "\x00" + strconv.FormatUint(gen, 10) + "\x00" +
104 + strconv.FormatInt(int64(grace), 10) + "\x00" + key
105 + ch := c.group.DoChan(flightKey, func() (any, error) {
106 + return c.refresh(ctx, key, epoch, gen, refresh, grace, logger, lookup)
107 + })
108 +
109 + select {
110 + case res := <-ch:
111 + if res.Err != nil {
112 + return nil, res.Err
113 + }
114 + return res.Val.([]Target), nil
115 + case <-ctx.Done():
116 + return nil, ctx.Err()
117 + }
118 +}
119 +
120 +// refresh runs a lookup for key on behalf of every caller sharing the flight.
121 +// epoch and gen are what the caller that started it observed.
122 +func (c *cache) refresh(ctx context.Context, key string, epoch, gen uint64, refresh, grace time.Duration, logger *zap.Logger, lookup func(context.Context) ([]Target, error)) ([]Target, error) {
123 + // a previous flight may have refreshed the entry since the caller looked
124 + c.mu.RLock()
125 + cached := c.entries[key]
126 + c.mu.RUnlock()
127 + if cached.isFresh() {
128 + return cached.targets, nil
129 + }
130 +
131 + // detached so the caller that started the flight can be cancelled without
132 + // failing it for the others, but bounded so a stuck resolver cannot
133 + // outlive every waiter
134 + ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), lookupTimeout)
135 + defer cancel()
136 +
137 + if err := lookups.acquire(ctx); err != nil {
138 + // slightly stale beats failing, but only if there is something to
139 + // serve; caching an empty result under load would be worse
140 + if grace > 0 && len(cached.targets) > 0 {
141 + if ce := logger.Check(zapcore.WarnLevel, c.kind+" lookup not started; using previously cached"); ce != nil {
142 + ce.Write(zap.Error(err))
143 + }
144 + return c.keepStale(key, epoch, gen, cached, refresh, grace), nil
145 + }
146 + return nil, err
147 + }
148 + targets, err := lookup(ctx)
149 + lookups.release()
150 + if err != nil {
151 + if grace > 0 {
152 + if ce := logger.Check(zapcore.ErrorLevel, c.kind+" lookup failed; using previously cached"); ce != nil {
153 + ce.Write(zap.Error(err))
154 + }
155 + return c.keepStale(key, epoch, gen, cached, refresh, grace), nil
156 + }
157 + return nil, err
158 + }
159 +
160 + c.store(key, epoch, gen, cacheEntry{refresh: refresh, freshness: time.Now(), targets: targets})
161 + return targets, nil
162 +}
163 +
164 +// keepStale extends cached so it is served for up to grace past its refresh,
165 +// and returns its targets.
166 +func (c *cache) keepStale(key string, epoch, gen uint64, cached cacheEntry, refresh, grace time.Duration) []Target {
167 + cached.freshness = time.Now().Add(grace - refresh)
168 + c.store(key, epoch, gen, cached)
169 + return cached.targets
170 +}
171 +
172 +// store writes e for key, unless key or the whole cache was reset since a
173 +// lookup observed epoch and gen.
174 +func (c *cache) store(key string, epoch, gen uint64, e cacheEntry) {
175 + c.mu.Lock()
176 + defer c.mu.Unlock()
177 + if c.epoch != epoch || c.entries[key].gen != gen {
178 + return
179 + }
180 + e.gen = gen
181 + c.makeRoom(key)
182 + c.entries[key] = e
183 +}
184 +
185 +// reset invalidates key. The entry is kept with its targets dropped and gen
186 +// bumped, rather than deleted, so a lookup for it that is in flight has a
187 +// generation to compare against.
188 +func (c *cache) reset(key string) {
189 + c.mu.Lock()
190 + defer c.mu.Unlock()
191 + // Keep the reset effective if cache pressure evicts its per-key marker.
192 + c.epoch++
193 + c.makeRoom(key)
194 + c.entries[key] = cacheEntry{gen: c.entries[key].gen + 1}
195 +}
196 +
197 +// resetAll drops every entry. Bumping epoch covers lookups in flight, so the
198 +// entries themselves can go.
199 +func (c *cache) resetAll() {
200 + c.mu.Lock()
201 + defer c.mu.Unlock()
202 + c.entries = make(map[string]cacheEntry)
203 + c.epoch++
204 +}
205 +
206 +// makeRoom evicts a random entry if the cache is full and key is not already
207 +// in it. c.mu must be held.
208 +func (c *cache) makeRoom(key string) {
209 + if _, ok := c.entries[key]; ok || len(c.entries) < maxCacheEntries {
210 + return
211 + }
212 + for k := range c.entries {
213 + delete(c.entries, k)
214 + break
215 + }
216 +}
217 +
218 +// lookupGate is a counting semaphore with a bounded wait queue.
219 +type lookupGate struct {
220 + running chan struct{}
221 + pending atomic.Int64
222 + maxPending int64
223 +}
224 +
225 +func newLookupGate(maxRunning, maxPending int) *lookupGate {
226 + return &lookupGate{
227 + running: make(chan struct{}, maxRunning),
228 + maxPending: int64(maxPending),
229 + }
230 +}
231 +
232 +// acquire takes a slot, waiting until ctx is done. It fails immediately with
233 +// errTooManyLookups if the wait queue is full. A free slot goes to whoever asks
234 +// first, even ahead of queued callers; that keeps the common case to a single
235 +// channel send.
236 +func (g *lookupGate) acquire(ctx context.Context) error {
237 + select {
238 + case g.running <- struct{}{}:
239 + return nil
240 + default:
241 + }
242 + if g.pending.Add(1) > g.maxPending {
243 + g.pending.Add(-1)
244 + return errTooManyLookups
245 + }
246 + defer g.pending.Add(-1)
247 + select {
248 + case g.running <- struct{}{}:
249 + return nil
250 + case <-ctx.Done():
251 + return ctx.Err()
252 + }
253 +}
254 +
255 +// release gives back a slot taken by a successful acquire.
256 +func (g *lookupGate) release() {
257 + <-g.running
258 +}
modules/internal/network/cache_test.go new
+158
@@ -0,0 +1,158 @@
1 +// Copyright 2015 Matthew Holt and The Caddy Authors
2 +//
3 +// Licensed under the Apache License, Version 2.0 (the "License");
4 +// you may not use this file except in compliance with the License.
5 +// You may obtain a copy of the License at
6 +//
7 +// http://www.apache.org/licenses/LICENSE-2.0
8 +//
9 +// Unless required by applicable law or agreed to in writing, software
10 +// distributed under the License is distributed on an "AS IS" BASIS,
11 +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 +// See the License for the specific language governing permissions and
13 +// limitations under the License.
14 +
15 +package network
16 +
17 +import (
18 + "context"
19 + "errors"
20 + "testing"
21 + "time"
22 +
23 + "go.uber.org/zap"
24 +)
25 +
26 +func TestLookupGate(t *testing.T) {
27 + g := newLookupGate(2, 1)
28 + for i := range 2 {
29 + if err := g.acquire(context.Background()); err != nil {
30 + t.Fatalf("acquire %d: %v", i, err)
31 + }
32 + }
33 +
34 + // the third caller queues; with the queue then full, a fourth is shed
35 + queued := make(chan error, 1)
36 + queuedCtx, cancelQueued := context.WithCancel(context.Background())
37 + go func() { queued <- g.acquire(queuedCtx) }()
38 + for g.pending.Load() != 1 {
39 + time.Sleep(time.Millisecond)
40 + }
41 + if err := g.acquire(context.Background()); !errors.Is(err, errTooManyLookups) {
42 + t.Fatalf("acquire with full queue: err = %v, want errTooManyLookups", err)
43 + }
44 +
45 + // a queued caller can give up
46 + cancelQueued()
47 + if err := <-queued; !errors.Is(err, context.Canceled) {
48 + t.Fatalf("cancelled queued acquire: err = %v, want context.Canceled", err)
49 + }
50 +
51 + // and a released slot is usable again
52 + g.release()
53 + if err := g.acquire(context.Background()); err != nil {
54 + t.Fatalf("acquire after release: %v", err)
55 + }
56 +}
57 +
58 +func TestEvictedResetMarkerRejectsStaleLookup(t *testing.T) {
59 + c := newCache("test")
60 + started := make(chan struct{})
61 + release := make(chan struct{})
62 + done := make(chan error, 1)
63 +
64 + go func() {
65 + _, err := c.get(context.Background(), "key", time.Minute, 0, zap.NewNop(), func(context.Context) ([]Target, error) {
66 + close(started)
67 + <-release
68 + return []Target{{Host: "old.example"}}, nil
69 + })
70 + done <- err
71 + }()
72 + <-started
73 +
74 + c.reset("key")
75 + c.mu.Lock()
76 + delete(c.entries, "key")
77 + c.mu.Unlock()
78 + close(release)
79 +
80 + if err := <-done; err != nil {
81 + t.Fatal(err)
82 + }
83 + c.mu.RLock()
84 + _, exists := c.entries["key"]
85 + c.mu.RUnlock()
86 + if exists {
87 + t.Fatal("lookup repopulated an entry after its reset marker was evicted")
88 + }
89 +}
90 +
91 +func TestDifferentGracePoliciesDoNotShareLookup(t *testing.T) {
92 + for _, tc := range []struct {
93 + name string
94 + graceFirst bool
95 + }{
96 + {name: "no grace first"},
97 + {name: "grace first", graceFirst: true},
98 + } {
99 + t.Run(tc.name, func(t *testing.T) {
100 + c := newCache("test")
101 + c.entries["key"] = cacheEntry{
102 + refresh: time.Minute,
103 + freshness: time.Now().Add(-time.Hour),
104 + targets: []Target{{Host: "stale.example"}},
105 + }
106 +
107 + started := make(chan struct{}, 2)
108 + release := make(chan struct{})
109 + lookup := func(context.Context) ([]Target, error) {
110 + started <- struct{}{}
111 + <-release
112 + return nil, errors.New("lookup failed")
113 + }
114 + type result struct {
115 + targets []Target
116 + err error
117 + }
118 + start := func(grace time.Duration) <-chan result {
119 + ch := make(chan result, 1)
120 + go func() {
121 + targets, err := c.get(context.Background(), "key", time.Minute, grace, zap.NewNop(), lookup)
122 + ch <- result{targets, err}
123 + }()
124 + return ch
125 + }
126 +
127 + firstGrace := time.Duration(0)
128 + secondGrace := time.Minute
129 + if tc.graceFirst {
130 + firstGrace, secondGrace = secondGrace, firstGrace
131 + }
132 + first := start(firstGrace)
133 + <-started
134 + second := start(secondGrace)
135 + select {
136 + case <-started:
137 + case <-time.After(time.Second):
138 + close(release)
139 + <-first
140 + <-second
141 + t.Fatalf("different grace policies shared one lookup")
142 + }
143 + close(release)
144 +
145 + firstResult := <-first
146 + secondResult := <-second
147 + if tc.graceFirst {
148 + firstResult, secondResult = secondResult, firstResult
149 + }
150 + if firstResult.err == nil {
151 + t.Fatal("caller without grace did not receive the lookup error")
152 + }
153 + if secondResult.err != nil || len(secondResult.targets) != 1 || secondResult.targets[0].Host != "stale.example" {
154 + t.Fatalf("caller with grace: targets = %v, err = %v", secondResult.targets, secondResult.err)
155 + }
156 + })
157 + }
158 +}
modules/internal/network/dynamicupstreams.go
+78 -126
@@ -26,7 +26,6 @@ import (
26 "context"
27 "net"
28 "strconv"
29 - "sync"
29 "time"
30
31 "go.uber.org/zap"
@@ -59,107 +58,66 @@ type SRVLookupFunc func(ctx context.Context, service, proto, name string) (cname
58 // past its refresh instead of returning an error. With no cached result, grace
59 // also suppresses the error and returns no targets, preserving the reverse
60 // proxy's existing behavior.
61 +//
62 +// Concurrent calls for the same record share one lookup, and no lock is held
63 +// while it runs, so a slow lookup does not delay calls for other records. The
64 +// shared lookup is detached from ctx (and bounded by its own timeout), so the
65 +// caller that started it can go away without failing it for the others; ctx
66 +// still bounds how long this call waits for the result.
67 func SRV(ctx context.Context, lookup SRVLookupFunc, resolverKey, service, proto, name string, refresh, grace time.Duration, logger *zap.Logger) ([]Target, error) {
68 key := srvKey(resolverKey, service, proto, name)
64 -
65 - // fast path: a fresh cached result under a read lock
66 - srvMu.RLock()
67 - cached := srvCache[key]
68 - srvMu.RUnlock()
69 - if cached.isFresh() {
70 - return cached.targets, nil
71 - }
72 -
73 - srvMu.Lock()
74 - defer srvMu.Unlock()
75 -
76 - // re-check under the write lock in case another goroutine refreshed it
77 - cached = srvCache[key]
78 - if cached.isFresh() {
79 - return cached.targets, nil
80 - }
81 -
82 - if c := logger.Check(zapcore.DebugLevel, "refreshing SRV upstreams"); c != nil {
83 - c.Write(zap.String("service", service), zap.String("proto", proto), zap.String("name", name))
84 - }
85 -
86 - _, records, err := lookup(ctx, service, proto, name)
87 - if err != nil {
88 - // From LookupSRV docs: invalid names are filtered out and an error is
89 - // returned alongside any remaining results; only treat it as fatal when
90 - // nothing usable came back.
91 - if len(records) == 0 {
92 - if grace > 0 {
93 - if c := logger.Check(zapcore.ErrorLevel, "SRV lookup failed; using previously cached"); c != nil {
94 - c.Write(zap.Error(err))
95 - }
96 - cached.freshness = time.Now().Add(grace - refresh)
97 - srvCache[key] = cached
98 - return cached.targets, nil
99 - }
100 - return nil, err
69 + return srvCache.get(ctx, key, refresh, grace, logger, func(ctx context.Context) ([]Target, error) {
70 + if c := logger.Check(zapcore.DebugLevel, "refreshing SRV upstreams"); c != nil {
71 + c.Write(zap.String("service", service), zap.String("proto", proto), zap.String("name", name))
72 }
102 - if c := logger.Check(zapcore.WarnLevel, "SRV records filtered"); c != nil {
103 - c.Write(zap.Error(err))
104 - }
105 - }
73
107 - targets := make([]Target, len(records))
108 - for i, rec := range records {
109 - if c := logger.Check(zapcore.DebugLevel, "discovered SRV record"); c != nil {
110 - c.Write(zap.String("target", rec.Target), zap.Uint16("port", rec.Port),
111 - zap.Uint16("priority", rec.Priority), zap.Uint16("weight", rec.Weight))
112 - }
113 - targets[i] = Target{
114 - Host: rec.Target,
115 - Port: strconv.Itoa(int(rec.Port)),
116 - Priority: rec.Priority,
117 - Weight: rec.Weight,
74 + _, records, err := lookup(ctx, service, proto, name)
75 + if err != nil {
76 + // From LookupSRV docs: invalid names are filtered out and an error is
77 + // returned alongside any remaining results; only treat it as fatal when
78 + // nothing usable came back.
79 + if len(records) == 0 {
80 + return nil, err
81 + }
82 + if c := logger.Check(zapcore.WarnLevel, "SRV records filtered"); c != nil {
83 + c.Write(zap.Error(err))
84 + }
85 }
119 - }
86
121 - // when inserting a brand-new entry (not replacing a stale one), bound the cache
122 - if cached.freshness.IsZero() && len(srvCache) >= maxCacheEntries {
123 - for k := range srvCache {
124 - delete(srvCache, k)
125 - break
87 + targets := make([]Target, len(records))
88 + for i, rec := range records {
89 + if c := logger.Check(zapcore.DebugLevel, "discovered SRV record"); c != nil {
90 + c.Write(zap.String("target", rec.Target), zap.Uint16("port", rec.Port),
91 + zap.Uint16("priority", rec.Priority), zap.Uint16("weight", rec.Weight))
92 + }
93 + targets[i] = Target{
94 + Host: rec.Target,
95 + Port: strconv.Itoa(int(rec.Port)),
96 + Priority: rec.Priority,
97 + Weight: rec.Weight,
98 + }
99 }
127 - }
128 - srvCache[key] = cacheEntry{refresh: refresh, freshness: time.Now(), targets: targets}
129 - return targets, nil
100 + return targets, nil
101 + })
102 }
103
104 // ResetSRV removes the cached result for a single SRV record resolved with the
133 -// given resolver (see [SRV] for resolverKey).
105 +// given resolver (see [SRV] for resolverKey). A lookup for it that is already
106 +// in flight will not write its result back.
107 func ResetSRV(resolverKey, service, proto, name string) {
135 - srvMu.Lock()
136 - delete(srvCache, srvKey(resolverKey, service, proto, name))
137 - srvMu.Unlock()
108 + srvCache.reset(srvKey(resolverKey, service, proto, name))
109 }
110
140 -// ResetAllSRV clears the entire SRV cache.
111 +// ResetAllSRV clears the entire SRV cache. Lookups already in flight will not
112 +// write their results back.
113 func ResetAllSRV() {
142 - srvMu.Lock()
143 - srvCache = make(map[string]cacheEntry)
144 - srvMu.Unlock()
114 + srvCache.resetAll()
115 }
116
117 func srvKey(resolverKey, service, proto, name string) string {
118 return resolverKey + "\x00" + service + "\x00" + proto + "\x00" + name
119 }
120
151 -const maxCacheEntries = 100
152 -
153 -type cacheEntry struct {
154 - refresh time.Duration
155 - freshness time.Time
156 - targets []Target
157 -}
158 -
159 -func (e cacheEntry) isFresh() bool {
160 - return !e.freshness.IsZero() && time.Since(e.freshness) < e.refresh
161 -}
162 -
121 // IPLookupFunc resolves a host's IP addresses. It matches the signature of
122 // (*net.Resolver).LookupIP, so callers can pass that directly (and tests can
123 // inject a stub). network is one of "ip", "ip4", "ip6".
@@ -171,55 +129,49 @@ type IPLookupFunc func(ctx context.Context, network, host string) ([]net.IP, err
129 // (*net.Resolver).LookupIP). resolverKey identifies that resolver, so results
130 // looked up with different resolvers for the same name are cached separately;
131 // pass "" for the default resolver.
132 +//
133 +// Concurrency and cancellation behave as described for [SRV].
134 func A(ctx context.Context, lookup IPLookupFunc, resolverKey, network, name, port string, refresh time.Duration, logger *zap.Logger) ([]Target, error) {
175 - key := resolverKey + "\x00" + name + "\x00" + port + "\x00" + network
176 -
177 - aMu.RLock()
178 - cached := aCache[key]
179 - aMu.RUnlock()
180 - if cached.isFresh() {
181 - return cached.targets, nil
182 - }
183 -
184 - aMu.Lock()
185 - defer aMu.Unlock()
186 -
187 - cached = aCache[key]
188 - if cached.isFresh() {
189 - return cached.targets, nil
190 - }
191 -
192 - if c := logger.Check(zapcore.DebugLevel, "refreshing A upstreams"); c != nil {
193 - c.Write(zap.String("version", network), zap.String("name", name), zap.String("port", port))
194 - }
195 -
196 - ips, err := lookup(ctx, network, name)
197 - if err != nil {
198 - return nil, err
199 - }
200 -
201 - targets := make([]Target, len(ips))
202 - for i, ip := range ips {
203 - if c := logger.Check(zapcore.DebugLevel, "discovered A record"); c != nil {
204 - c.Write(zap.String("ip", ip.String()))
135 + key := aKey(resolverKey, network, name, port)
136 + return aCache.get(ctx, key, refresh, 0, logger, func(ctx context.Context) ([]Target, error) {
137 + if c := logger.Check(zapcore.DebugLevel, "refreshing A upstreams"); c != nil {
138 + c.Write(zap.String("version", network), zap.String("name", name), zap.String("port", port))
139 }
206 - targets[i] = Target{Host: ip.String(), Port: port}
207 - }
140
209 - if cached.freshness.IsZero() && len(aCache) >= maxCacheEntries {
210 - for k := range aCache {
211 - delete(aCache, k)
212 - break
141 + ips, err := lookup(ctx, network, name)
142 + if err != nil {
143 + return nil, err
144 }
214 - }
215 - aCache[key] = cacheEntry{refresh: refresh, freshness: time.Now(), targets: targets}
216 - return targets, nil
145 +
146 + targets := make([]Target, len(ips))
147 + for i, ip := range ips {
148 + if c := logger.Check(zapcore.DebugLevel, "discovered A record"); c != nil {
149 + c.Write(zap.String("ip", ip.String()))
150 + }
151 + targets[i] = Target{Host: ip.String(), Port: port}
152 + }
153 + return targets, nil
154 + })
155 }
156
219 -var (
220 - srvMu sync.RWMutex
221 - srvCache = make(map[string]cacheEntry)
157 +// ResetA removes the cached result for a single A/AAAA lookup (see [A] for the
158 +// parameters). A lookup for it that is already in flight will not write its
159 +// result back.
160 +func ResetA(resolverKey, network, name, port string) {
161 + aCache.reset(aKey(resolverKey, network, name, port))
162 +}
163 +
164 +// ResetAllA clears the entire A/AAAA cache. Lookups already in flight will not
165 +// write their results back.
166 +func ResetAllA() {
167 + aCache.resetAll()
168 +}
169
223 - aMu sync.RWMutex
224 - aCache = make(map[string]cacheEntry)
170 +func aKey(resolverKey, network, name, port string) string {
171 + return resolverKey + "\x00" + name + "\x00" + port + "\x00" + network
172 +}
173 +
174 +var (
175 + srvCache = newCache("SRV")
176 + aCache = newCache("A")
177 )
modules/internal/network/dynamicupstreams_test.go
+300 -44
@@ -20,6 +20,7 @@ import (
20 "fmt"
21 "net"
22 "sync"
23 + "sync/atomic"
24 "testing"
25 "time"
26
@@ -168,18 +169,18 @@ func TestResetAllSRV(t *testing.T) {
169 if _, err := SRV(context.Background(), lookup, "", "svc-reset-all", "tcp", "x", time.Minute, 0, zap.NewNop()); err != nil {
170 t.Fatalf("SRV: %v", err)
171 }
171 - srvMu.RLock()
172 - populated := len(srvCache) > 0
173 - srvMu.RUnlock()
172 + srvCache.mu.RLock()
173 + populated := len(srvCache.entries) > 0
174 + srvCache.mu.RUnlock()
175 if !populated {
176 t.Fatal("expected the SRV cache to be populated before reset")
177 }
178
179 ResetAllSRV()
180
180 - srvMu.RLock()
181 - n := len(srvCache)
182 - srvMu.RUnlock()
181 + srvCache.mu.RLock()
182 + n := len(srvCache.entries)
183 + srvCache.mu.RUnlock()
184 if n != 0 {
185 t.Fatalf("srvCache len = %d after ResetAllSRV, want 0", n)
186 }
@@ -188,9 +189,7 @@ func TestResetAllSRV(t *testing.T) {
189 // TestSRVCacheBound verifies that inserting a brand-new entry once the cache is
190 // full evicts an existing one so the cache stays bounded.
191 func TestSRVCacheBound(t *testing.T) {
191 - srvMu.Lock()
192 - srvCache = make(map[string]cacheEntry)
193 - srvMu.Unlock()
192 + ResetAllSRV()
193
194 lookup := func(context.Context, string, string, string) (string, []*net.SRV, error) {
195 return "", []*net.SRV{{Target: "a.example.", Port: 1}}, nil
@@ -201,9 +200,9 @@ func TestSRVCacheBound(t *testing.T) {
200 t.Fatalf("SRV[%d]: %v", i, err)
201 }
202 }
204 - srvMu.RLock()
205 - n := len(srvCache)
206 - srvMu.RUnlock()
203 + srvCache.mu.RLock()
204 + n := len(srvCache.entries)
205 + srvCache.mu.RUnlock()
206 if n > maxCacheEntries {
207 t.Fatalf("srvCache len = %d, want <= %d (cache must stay bounded)", n, maxCacheEntries)
208 }
@@ -240,9 +239,7 @@ func TestSRVResolverKeyIsolatesCache(t *testing.T) {
239
240 // TestAResolverKeyIsolatesCache is the A-cache equivalent of the above.
241 func TestAResolverKeyIsolatesCache(t *testing.T) {
243 - aMu.Lock()
244 - aCache = make(map[string]cacheEntry)
245 - aMu.Unlock()
242 + ResetAllA()
243
244 lookupA := func(context.Context, string, string) ([]net.IP, error) {
245 return []net.IP{net.ParseIP("10.0.0.1")}, nil
@@ -269,9 +266,7 @@ func TestAResolverKeyIsolatesCache(t *testing.T) {
266
267 // TestACacheBound verifies the same bounding behavior for the A cache.
268 func TestACacheBound(t *testing.T) {
272 - aMu.Lock()
273 - aCache = make(map[string]cacheEntry)
274 - aMu.Unlock()
269 + ResetAllA()
270
271 lookup := func(context.Context, string, string) ([]net.IP, error) {
272 return []net.IP{net.ParseIP("10.0.0.1")}, nil
@@ -282,26 +277,22 @@ func TestACacheBound(t *testing.T) {
277 t.Fatalf("A[%d]: %v", i, err)
278 }
279 }
285 - aMu.RLock()
286 - n := len(aCache)
287 - aMu.RUnlock()
280 + aCache.mu.RLock()
281 + n := len(aCache.entries)
282 + aCache.mu.RUnlock()
283 if n > maxCacheEntries {
284 t.Fatalf("aCache len = %d, want <= %d (cache must stay bounded)", n, maxCacheEntries)
285 }
286 }
287
288 // TestSRVConcurrentRefreshDeduplicates verifies that two concurrent callers
294 -// missing the cache for the same key result in a single lookup: the first
295 -// performs it while holding the write lock, and the second is served the
296 -// cached result once the first releases. Depending on scheduling the second
297 -// caller may be served by either the read-lock fast path or the write-lock
298 -// re-check — this test does not force one branch over the other; it asserts
299 -// deduplication (exactly one lookup). Run under -race it also guards against
289 +// missing the cache for the same key result in a single lookup. Depending on
290 +// scheduling the second caller either joins the first one's in-flight lookup
291 +// or is served the result it cached; this test does not force one over the
292 +// other, it asserts deduplication (exactly one lookup). Run under -race it also guards against
293 // data races on the shared cache.
294 func TestSRVConcurrentRefreshDeduplicates(t *testing.T) {
302 - srvMu.Lock()
303 - srvCache = make(map[string]cacheEntry)
304 - srvMu.Unlock()
295 + ResetAllSRV()
296
297 var calls int
298 var mu sync.Mutex
@@ -314,24 +305,21 @@ func TestSRVConcurrentRefreshDeduplicates(t *testing.T) {
305 mu.Unlock()
306 if first {
307 close(inLookup)
317 - <-release // hold the write lock until the second goroutine is queued
308 + <-release // hold the lookup open while the second goroutine starts
309 }
310 return "", []*net.SRV{{Target: "a.example.", Port: 1}}, nil
311 }
312
313 done := make(chan struct{}, 2)
323 - // G1: takes the write lock and blocks inside lookup.
314 + // G1: starts the lookup and blocks inside it.
315 go func() {
316 _, _ = SRV(context.Background(), lookup, "", "svc-conc", "tcp", "x", time.Minute, 0, zap.NewNop())
317 done <- struct{}{}
318 }()
328 - <-inLookup // G1 now holds the write lock; cache is still empty
319 + <-inLookup // G1 is mid-lookup; cache is still empty
320
330 - // G2: a second caller for the same key. While G1 holds the write lock, G2
331 - // cannot acquire srvMu at all (a reader blocks behind the writer); once G1
332 - // populates the cache and releases, G2 observes the cached entry and
333 - // performs no lookup. Ordering here is enforced by the channels below, not
334 - // by timing.
321 + // G2: a second caller for the same key; it must not start a lookup of its
322 + // own.
323 go func() {
324 _, _ = SRV(context.Background(), lookup, "", "svc-conc", "tcp", "x", time.Minute, 0, zap.NewNop())
325 done <- struct{}{}
@@ -353,9 +341,7 @@ func TestSRVConcurrentRefreshDeduplicates(t *testing.T) {
341 // it asserts deduplication (one lookup for two concurrent callers) and, under
342 // -race, race-safety; it does not force a particular locking branch.
343 func TestAConcurrentRefreshDeduplicates(t *testing.T) {
356 - aMu.Lock()
357 - aCache = make(map[string]cacheEntry)
358 - aMu.Unlock()
344 + ResetAllA()
345
346 var calls int
347 var mu sync.Mutex
@@ -438,9 +424,7 @@ func TestDiscoveryLoggingOnlyOnRefresh(t *testing.T) {
424 }
425 })
426 t.Run("A", func(t *testing.T) {
441 - aMu.Lock()
442 - aCache = make(map[string]cacheEntry)
443 - aMu.Unlock()
427 + ResetAllA()
428 core, logs := observer.New(zap.DebugLevel)
429 logger := zap.New(core)
430 lookup := func(context.Context, string, string) ([]net.IP, error) {
@@ -463,3 +447,275 @@ func TestDiscoveryLoggingOnlyOnRefresh(t *testing.T) {
447 }
448 })
449 }
450 +
451 +// blockingSRV returns an SRV lookup that sends its context on started and then
452 +// blocks until release is closed, answering with target.
453 +func blockingSRV(target string, started chan<- context.Context, release <-chan struct{}) SRVLookupFunc {
454 + return func(ctx context.Context, _, _, _ string) (string, []*net.SRV, error) {
455 + started <- ctx
456 + <-release
457 + return "", []*net.SRV{{Target: target, Port: 1}}, nil
458 + }
459 +}
460 +
461 +// instantSRV returns an SRV lookup that answers with target right away,
462 +// counting its calls.
463 +func instantSRV(target string, calls *atomic.Int32) SRVLookupFunc {
464 + return func(context.Context, string, string, string) (string, []*net.SRV, error) {
465 + calls.Add(1)
466 + return "", []*net.SRV{{Target: target, Port: 1}}, nil
467 + }
468 +}
469 +
470 +type srvResult struct {
471 + targets []Target
472 + err error
473 +}
474 +
475 +func goSRV(ctx context.Context, lookup SRVLookupFunc, service string) <-chan srvResult {
476 + ch := make(chan srvResult, 1)
477 + go func() {
478 + targets, err := SRV(ctx, lookup, "", service, "tcp", "x", time.Minute, 0, zap.NewNop())
479 + ch <- srvResult{targets, err}
480 + }()
481 + return ch
482 +}
483 +
484 +func waitResult(t *testing.T, ch <-chan srvResult) srvResult {
485 + t.Helper()
486 + select {
487 + case res := <-ch:
488 + return res
489 + case <-time.After(5 * time.Second):
490 + t.Fatal("timed out waiting for SRV to return")
491 + return srvResult{}
492 + }
493 +}
494 +
495 +// TestSlowLookupDoesNotBlockOtherKeys is a regression test for #7946: a slow
496 +// lookup for one key used to hold the cache lock across the DNS call, blocking
497 +// every other key.
498 +func TestSlowLookupDoesNotBlockOtherKeys(t *testing.T) {
499 + started := make(chan context.Context, 1)
500 + release := make(chan struct{})
501 + slow := goSRV(context.Background(), blockingSRV("slow.example.", started, release), "svc-slow")
502 + <-started
503 +
504 + var calls atomic.Int32
505 + res := waitResult(t, goSRV(context.Background(), instantSRV("fast.example.", &calls), "svc-fast"))
506 + if res.err != nil || len(res.targets) != 1 || res.targets[0].Host != "fast.example." {
507 + t.Fatalf("unrelated key while another is slow: targets = %v, err = %v", res.targets, res.err)
508 + }
509 +
510 + close(release)
511 + if res := waitResult(t, slow); res.err != nil {
512 + t.Fatalf("slow lookup: %v", res.err)
513 + }
514 +}
515 +
516 +// TestCallerCancellationDoesNotFailSharedLookup checks that a shared lookup
517 +// outlives the caller that started it: that caller returns promptly with its
518 +// own context error, while another caller on the same flight still gets the
519 +// result. The lookup must still be bounded by a deadline of its own.
520 +func TestCallerCancellationDoesNotFailSharedLookup(t *testing.T) {
521 + started := make(chan context.Context, 2)
522 + release := make(chan struct{})
523 + lookup := blockingSRV("a.example.", started, release)
524 +
525 + leaderCtx, cancelLeader := context.WithCancel(context.Background())
526 + leader := goSRV(leaderCtx, lookup, "svc-leader")
527 + lookupCtx := <-started
528 + follower := goSRV(context.Background(), lookup, "svc-leader")
529 +
530 + cancelLeader()
531 + if res := waitResult(t, leader); !errors.Is(res.err, context.Canceled) {
532 + t.Fatalf("cancelled caller: err = %v, want context.Canceled", res.err)
533 + }
534 + if err := lookupCtx.Err(); err != nil {
535 + t.Fatalf("shared lookup was cancelled with its first caller: %v", err)
536 + }
537 + if _, ok := lookupCtx.Deadline(); !ok {
538 + t.Fatal("shared lookup has no deadline of its own")
539 + }
540 +
541 + close(release)
542 + res := waitResult(t, follower)
543 + if res.err != nil || len(res.targets) != 1 {
544 + t.Fatalf("remaining caller: targets = %v, err = %v", res.targets, res.err)
545 + }
546 + select {
547 + case <-started:
548 + t.Fatal("remaining caller started a second lookup instead of sharing the first")
549 + default:
550 + }
551 +}
552 +
553 +// TestResetDuringLookup checks that a reset invalidates a lookup already in
554 +// flight: a caller arriving after the reset starts a new lookup rather than
555 +// joining the old one, and the old one, finishing last, does not overwrite the
556 +// newer result.
557 +func TestResetDuringLookup(t *testing.T) {
558 + started := make(chan context.Context, 1)
559 + release := make(chan struct{})
560 + old := goSRV(context.Background(), blockingSRV("old.example.", started, release), "svc-reset-inflight")
561 + <-started
562 +
563 + ResetSRV("", "svc-reset-inflight", "tcp", "x")
564 +
565 + var calls atomic.Int32
566 + res := waitResult(t, goSRV(context.Background(), instantSRV("new.example.", &calls), "svc-reset-inflight"))
567 + if res.err != nil || calls.Load() != 1 || res.targets[0].Host != "new.example." {
568 + t.Fatalf("caller after reset: targets = %v, err = %v, lookups = %d; want its own lookup", res.targets, res.err, calls.Load())
569 + }
570 +
571 + close(release)
572 + if res := waitResult(t, old); res.err != nil {
573 + t.Fatalf("pre-reset lookup: %v", res.err)
574 + }
575 +
576 + res = waitResult(t, goSRV(context.Background(), instantSRV("unused.example.", &calls), "svc-reset-inflight"))
577 + if res.err != nil || res.targets[0].Host != "new.example." {
578 + t.Fatalf("cached after both lookups: targets = %v, err = %v; want new.example.", res.targets, res.err)
579 + }
580 +}
581 +
582 +// TestResetAllDuringLookup checks that a full reset also keeps an in-flight
583 +// lookup from repopulating the cache.
584 +func TestResetAllDuringLookup(t *testing.T) {
585 + started := make(chan context.Context, 1)
586 + release := make(chan struct{})
587 + old := goSRV(context.Background(), blockingSRV("old.example.", started, release), "svc-reset-all-inflight")
588 + <-started
589 +
590 + ResetAllSRV()
591 + close(release)
592 + if res := waitResult(t, old); res.err != nil {
593 + t.Fatalf("pre-reset lookup: %v", res.err)
594 + }
595 +
596 + srvCache.mu.RLock()
597 + _, ok := srvCache.entries[srvKey("", "svc-reset-all-inflight", "tcp", "x")]
598 + srvCache.mu.RUnlock()
599 + if ok {
600 + t.Fatal("lookup that started before ResetAllSRV wrote its result back")
601 + }
602 +}
603 +
604 +// saturateLookups fills every slot of the shared lookup gate, and optionally
605 +// its wait queue too, until the test ends.
606 +func saturateLookups(t *testing.T, fillQueue bool) {
607 + t.Helper()
608 + for range cap(lookups.running) {
609 + lookups.running <- struct{}{}
610 + }
611 + if fillQueue {
612 + lookups.pending.Add(lookups.maxPending)
613 + }
614 + t.Cleanup(func() {
615 + if fillQueue {
616 + lookups.pending.Add(-lookups.maxPending)
617 + }
618 + for range cap(lookups.running) {
619 + lookups.release()
620 + }
621 + })
622 +}
623 +
624 +// TestLookupsAreBounded checks the gate is wired in: while it is saturated a
625 +// new key does not reach the resolver and its caller can still give up, and
626 +// the lookup proceeds once slots free up.
627 +func TestLookupsAreBounded(t *testing.T) {
628 + var calls atomic.Int32
629 + lookup := instantSRV("a.example.", &calls)
630 +
631 + t.Run("saturated", func(t *testing.T) {
632 + saturateLookups(t, false)
633 + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
634 + defer cancel()
635 + if res := waitResult(t, goSRV(ctx, lookup, "svc-bounded")); !errors.Is(res.err, context.DeadlineExceeded) {
636 + t.Fatalf("caller while saturated: err = %v, want its own deadline", res.err)
637 + }
638 + if n := calls.Load(); n != 0 {
639 + t.Fatalf("resolver called %d times while the gate was saturated", n)
640 + }
641 + })
642 +
643 + res := waitResult(t, goSRV(context.Background(), lookup, "svc-bounded"))
644 + if res.err != nil || len(res.targets) != 1 {
645 + t.Fatalf("after slots freed: targets = %v, err = %v", res.targets, res.err)
646 + }
647 +}
648 +
649 +// TestShedLookups checks what happens when the gate's queue is full too: a
650 +// stale SRV entry is served if there is a grace period; otherwise the lookup
651 +// fails without reaching the resolver.
652 +func TestShedLookups(t *testing.T) {
653 + var calls atomic.Int32
654 + seed := instantSRV("stale.example.", &calls)
655 + for _, service := range []string{"svc-shed-stale", "svc-shed-nograce"} {
656 + if _, err := SRV(context.Background(), seed, "", service, "tcp", "x", time.Minute, 0, zap.NewNop()); err != nil {
657 + t.Fatal(err)
658 + }
659 + // backdate rather than relying on a tiny refresh, which a coarse
660 + // clock (e.g. on Windows) can still report as fresh
661 + key := srvKey("", service, "tcp", "x")
662 + srvCache.mu.Lock()
663 + e := srvCache.entries[key]
664 + e.freshness = time.Now().Add(-time.Hour)
665 + srvCache.entries[key] = e
666 + srvCache.mu.Unlock()
667 + }
668 + calls.Store(0)
669 +
670 + saturateLookups(t, true)
671 +
672 + targets, err := SRV(context.Background(), seed, "", "svc-shed-stale", "tcp", "x", time.Minute, time.Hour, zap.NewNop())
673 + if err != nil || len(targets) != 1 || targets[0].Host != "stale.example." {
674 + t.Fatalf("shed with grace and a stale entry: targets = %v, err = %v; want the stale entry", targets, err)
675 + }
676 + for _, tc := range []struct {
677 + name string
678 + grace time.Duration
679 + }{
680 + {name: "svc-shed-nograce"},
681 + {name: "svc-shed-cold", grace: time.Hour},
682 + } {
683 + if _, err := SRV(context.Background(), seed, "", tc.name, "tcp", "x", time.Minute, tc.grace, zap.NewNop()); !errors.Is(err, errTooManyLookups) {
684 + t.Errorf("%s: err = %v, want errTooManyLookups", tc.name, err)
685 + }
686 + }
687 + ipLookup := func(context.Context, string, string) ([]net.IP, error) {
688 + calls.Add(1)
689 + return []net.IP{net.ParseIP("10.0.0.1")}, nil
690 + }
691 + if _, err := A(context.Background(), ipLookup, "", "ip", "db.shed", "80", time.Minute, zap.NewNop()); !errors.Is(err, errTooManyLookups) {
692 + t.Errorf("A: err = %v, want errTooManyLookups", err)
693 + }
694 + if n := calls.Load(); n != 0 {
695 + t.Fatalf("resolver called %d times while lookups were being shed", n)
696 + }
697 +}
698 +
699 +func TestResetA(t *testing.T) {
700 + var calls atomic.Int32
701 + lookup := func(context.Context, string, string) ([]net.IP, error) {
702 + calls.Add(1)
703 + return []net.IP{net.ParseIP("10.0.0.1")}, nil
704 + }
705 + get := func() {
706 + t.Helper()
707 + if _, err := A(context.Background(), lookup, "", "ip", "db.reset", "80", time.Minute, zap.NewNop()); err != nil {
708 + t.Fatal(err)
709 + }
710 + }
711 +
712 + get()
713 + get()
714 + ResetA("", "ip", "db.reset", "80")
715 + get()
716 + ResetAllA()
717 + get()
718 + if n := calls.Load(); n != 3 {
719 + t.Fatalf("lookup calls = %d, want 3 (one, cached, then one after each reset)", n)
720 + }
721 +}