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, 03:30 UTC
reverseproxy: deduplicate DNS SRV+A discovery into modules/internal/network (#7790)
* reverseproxy: extract DNS SRV+A discovery into a reusable package Move the SRV and A/AAAA resolution + caching logic out of reverseproxy into a new, transport-neutral package (dynamicupstreams) that returns neutral targets instead of *reverseproxy.Upstream. reverseproxy's SRVUpstreams and AUpstreams now delegate to it and build their own upstreams from the targets; behavior is unchanged. This lets other proxies (e.g. third-party layer4 proxies) reuse the same DNS discovery + caching instead of copying it, reducing duplication and maintenance burden. reverseproxy/upstreams.go shrinks substantially. RFC for the de-duplication discussed in caddyserver/caddy-l4#429. * test(dynamicupstreams): cover ResetAllSRV, filtered records, cache bound, concurrent refresh Brings the package to 96.8% statement coverage. Adds: - TestResetAllSRV: full-cache drop - TestSRVFilteredRecords: LookupSRV partial-error semantics (usable records returned, error downgraded to a warning) - TestSRVCacheBound / TestACacheBound: eviction keeps the cache bounded at maxCacheEntries - TestSRVConcurrentRefreshDeduplicates / TestAConcurrentRefreshDeduplicates: two goroutines missing the read-lock fast path trigger only one lookup (verified under -race) The grace-period and filtered-records logs now use a level-enabled logger so the log-write branches execute. The only uncovered statements are the two double-checked-locking re-check returns, which are reachable solely in a narrow race window (the same untested idiom as reverseproxy's SRV/A). * reverseproxy: move DNS SRV+A discovery into modules/internal/network Per review feedback, the shared discovery code does not belong in a top-level package: it is internal helper code, not part of Caddy's runtime surface. Move it from the top-level dynamicupstreams package into the existing modules/internal/network package and update reverse_proxy's upstreams.go to consume it from there. Being under modules/internal means third-party modules (e.g. layer4) cannot import it yet; they keep their own copy until layer4 is part of Caddy's standard distribution, at which point they can switch over. * test(dynamicupstreams): correct the concurrent-refresh test's claim The two concurrent-refresh tests documented themselves as covering the write-lock re-check branch ("the second re-checks under the write lock"), but they do not: while the first goroutine holds the write lock the second cannot acquire the read lock either, so by the time it acquires srvMu the cache is already populated and it is served by the read-lock fast path. The 50ms sleep was presented as letting the second goroutine "queue on the write lock" but proved nothing. Make the comments and failure messages honest: these tests assert deduplication (two concurrent callers for one uncached key share a single lookup) and, under -race, race-safety on the shared cache -- not a specific locking branch. Ordering is enforced by the channels, so the sleep is removed. No production code change. * reverseproxy: preserve DNS discovery logging and grace behavior Keep refresh and discovery logs in the DNS lookup path, retaining the original SRV field types. Preserve error suppression when an SRV lookup fails with grace enabled before any result has been cached. Add regression tests for refresh-only logging and cold-cache grace handling. AI-assisted-by: OpenAI Codex (GPT-6) --------- Co-authored-by: Francis Lavoie <lavofr@gmail.com>
tannevaled committed
Oct 12, 2026 at 12:37 UTC
f2c20fb8107a5497d6276ed33f5a454c64329ce5
3 files changed
+647
-183
modules/caddyhttp/reverseproxy/upstreams.go
+21
-183
@@ -7,14 +7,13 @@ import (
7
weakrand "math/rand/v2"
8
"net"
9
"net/http"
10
- "strconv"
11
- "sync"
10
"time"
11
12
"go.uber.org/zap"
13
"go.uber.org/zap/zapcore"
14
15
"github.com/caddyserver/caddy/v2"
16
+ "github.com/caddyserver/caddy/v2/modules/internal/network"
17
)
18
19
func init() {
@@ -120,101 +119,34 @@ func (su *SRVUpstreams) Provision(ctx caddy.Context) error {
119
}
120
121
func (su *SRVUpstreams) ResetCache(r *http.Request) error {
123
- srvsMu.Lock()
122
if r == nil {
125
- srvs = make(map[string]srvLookup)
126
- } else {
127
- suAddr, _, _, _ := su.expandedAddr(r)
128
- delete(srvs, suAddr)
123
+ network.ResetAllSRV()
124
+ return nil
125
}
130
- srvsMu.Unlock()
126
+ _, service, proto, name := su.expandedAddr(r)
127
+ network.ResetSRV(service, proto, name)
128
return nil
129
}
130
131
func (su SRVUpstreams) GetUpstreams(r *http.Request) ([]*Upstream, error) {
135
- suAddr, service, proto, name := su.expandedAddr(r)
136
-
137
- // first, use a cheap read-lock to return a cached result quickly
138
- srvsMu.RLock()
139
- cached := srvs[suAddr]
140
- srvsMu.RUnlock()
141
- if cached.isFresh() {
142
- return allNew(cached.upstreams), nil
143
- }
144
-
145
- // otherwise, obtain a write-lock to update the cached value
146
- srvsMu.Lock()
147
- defer srvsMu.Unlock()
148
-
149
- // check to see if it's still stale, since we're now in a different
150
- // lock from when we first checked freshness; another goroutine might
151
- // have refreshed it in the meantime before we re-obtained our lock
152
- cached = srvs[suAddr]
153
- if cached.isFresh() {
154
- return allNew(cached.upstreams), nil
155
- }
132
+ _, service, proto, name := su.expandedAddr(r)
133
157
- if c := su.logger.Check(zapcore.DebugLevel, "refreshing SRV upstreams"); c != nil {
158
- c.Write(
159
- zap.String("service", service),
160
- zap.String("proto", proto),
161
- zap.String("name", name),
162
- )
163
- }
164
-
165
- _, records, err := su.resolver.LookupSRV(r.Context(), service, proto, name)
134
+ targets, err := network.SRV(r.Context(), su.resolver.LookupSRV,
135
+ service, proto, name,
136
+ time.Duration(su.Refresh), time.Duration(su.GracePeriod), su.logger)
137
if err != nil {
167
- // From LookupSRV docs: "If the response contains invalid names, those records are filtered
168
- // out and an error will be returned alongside the remaining results, if any." Thus, we
169
- // only return an error if no records were also returned.
170
- if len(records) == 0 {
171
- if su.GracePeriod > 0 {
172
- if c := su.logger.Check(zapcore.ErrorLevel, "SRV lookup failed; using previously cached"); c != nil {
173
- c.Write(zap.Error(err))
174
- }
175
- cached.freshness = time.Now().Add(time.Duration(su.GracePeriod) - time.Duration(su.Refresh))
176
- srvs[suAddr] = cached
177
- return allNew(cached.upstreams), nil
178
- }
179
- return nil, err
180
- }
181
- if c := su.logger.Check(zapcore.WarnLevel, "SRV records filtered"); c != nil {
182
- c.Write(zap.Error(err))
183
- }
138
+ return nil, err
139
}
140
186
- upstreams := make([]Upstream, len(records))
187
- for i, rec := range records {
188
- if c := su.logger.Check(zapcore.DebugLevel, "discovered SRV record"); c != nil {
189
- c.Write(
190
- zap.String("target", rec.Target),
191
- zap.Uint16("port", rec.Port),
192
- zap.Uint16("priority", rec.Priority),
193
- zap.Uint16("weight", rec.Weight),
194
- )
195
- }
196
- addr := net.JoinHostPort(rec.Target, strconv.Itoa(int(rec.Port)))
141
+ upstreams := make([]*Upstream, len(targets))
142
+ for i, t := range targets {
143
+ addr := net.JoinHostPort(t.Host, t.Port)
144
if su.DialNetwork != "" {
145
addr = su.DialNetwork + "/" + addr
146
}
200
- upstreams[i] = Upstream{Dial: addr}
201
- }
202
-
203
- // before adding a new one to the cache (as opposed to replacing stale one), make room if cache is full
204
- if cached.freshness.IsZero() && len(srvs) >= 100 {
205
- for randomKey := range srvs {
206
- delete(srvs, randomKey)
207
- break
208
- }
209
- }
210
-
211
- srvs[suAddr] = srvLookup{
212
- srvUpstreams: su,
213
- freshness: time.Now(),
214
- upstreams: upstreams,
147
+ upstreams[i] = &Upstream{Dial: addr}
148
}
216
-
217
- return allNew(upstreams), nil
149
+ return upstreams, nil
150
}
151
152
func (su SRVUpstreams) String() string {
@@ -247,16 +179,6 @@ func (SRVUpstreams) formattedAddr(service, proto, name string) string {
179
return fmt.Sprintf("_%s._%s.%s", service, proto, name)
180
}
181
250
-type srvLookup struct {
251
- srvUpstreams SRVUpstreams
252
- freshness time.Time
253
- upstreams []Upstream
254
-}
255
-
256
-func (sl srvLookup) isFresh() bool {
257
- return time.Since(sl.freshness) < time.Duration(sl.srvUpstreams.Refresh)
258
-}
259
-
182
type IPVersions struct {
183
IPv4 *bool `json:"ipv4,omitempty"`
184
IPv6 *bool `json:"ipv6,omitempty"`
@@ -357,93 +279,25 @@ func (au *AUpstreams) Provision(ctx caddy.Context) error {
279
func (au AUpstreams) GetUpstreams(r *http.Request) ([]*Upstream, error) {
280
repl := r.Context().Value(caddy.ReplacerCtxKey).(*caddy.Replacer)
281
360
- // Map ipVersion early, so we can use it as part of the cache-key.
361
- // This should be fairly inexpensive and comes and the upside of
362
- // allowing the same dynamic upstream (name + port combination)
363
- // to be used multiple times with different ip versions.
364
- //
365
- // It also forced a cache-miss if a previously cached dynamic
366
- // upstream changes its ip version, e.g. after a config reload,
367
- // while keeping the cache-invalidation as simple as it currently is.
282
ipVersion := resolveIpVersion(au.Versions)
369
-
370
- auStr := repl.ReplaceAll(au.String()+ipVersion, "")
371
-
372
- // first, use a cheap read-lock to return a cached result quickly
373
- aAaaaMu.RLock()
374
- cached := aAaaa[auStr]
375
- aAaaaMu.RUnlock()
376
- if cached.isFresh() {
377
- return allNew(cached.upstreams), nil
378
- }
379
-
380
- // otherwise, obtain a write-lock to update the cached value
381
- aAaaaMu.Lock()
382
- defer aAaaaMu.Unlock()
383
-
384
- // check to see if it's still stale, since we're now in a different
385
- // lock from when we first checked freshness; another goroutine might
386
- // have refreshed it in the meantime before we re-obtained our lock
387
- cached = aAaaa[auStr]
388
- if cached.isFresh() {
389
- return allNew(cached.upstreams), nil
390
- }
391
-
283
name := repl.ReplaceAll(au.Name, "")
284
port := repl.ReplaceAll(au.Port, "")
285
395
- if c := au.logger.Check(zapcore.DebugLevel, "refreshing A upstreams"); c != nil {
396
- c.Write(
397
- zap.String("version", ipVersion),
398
- zap.String("name", name),
399
- zap.String("port", port),
400
- )
401
- }
402
-
403
- ips, err := au.resolver.LookupIP(r.Context(), ipVersion, name)
286
+ targets, err := network.A(r.Context(), au.resolver.LookupIP,
287
+ ipVersion, name, port, time.Duration(au.Refresh), au.logger)
288
if err != nil {
289
return nil, err
290
}
291
408
- upstreams := make([]Upstream, len(ips))
409
- for i, ip := range ips {
410
- if c := au.logger.Check(zapcore.DebugLevel, "discovered A record"); c != nil {
411
- c.Write(zap.String("ip", ip.String()))
412
- }
413
- upstreams[i] = Upstream{
414
- Dial: net.JoinHostPort(ip.String(), port),
415
- }
292
+ upstreams := make([]*Upstream, len(targets))
293
+ for i, t := range targets {
294
+ upstreams[i] = &Upstream{Dial: net.JoinHostPort(t.Host, t.Port)}
295
}
417
-
418
- // before adding a new one to the cache (as opposed to replacing stale one), make room if cache is full
419
- if cached.freshness.IsZero() && len(aAaaa) >= 100 {
420
- for randomKey := range aAaaa {
421
- delete(aAaaa, randomKey)
422
- break
423
- }
424
- }
425
-
426
- aAaaa[auStr] = aLookup{
427
- aUpstreams: au,
428
- freshness: time.Now(),
429
- upstreams: upstreams,
430
- }
431
-
432
- return allNew(upstreams), nil
296
+ return upstreams, nil
297
}
298
299
func (au AUpstreams) String() string { return net.JoinHostPort(au.Name, au.Port) }
300
437
-type aLookup struct {
438
- aUpstreams AUpstreams
439
- freshness time.Time
440
- upstreams []Upstream
441
-}
442
-
443
-func (al aLookup) isFresh() bool {
444
- return time.Since(al.freshness) < time.Duration(al.aUpstreams.Refresh)
445
-}
446
-
301
// MultiUpstreams is a single dynamic upstream source that
302
// aggregates the results of multiple dynamic upstream sources.
303
// All configured sources will be queried in order, with their
@@ -551,22 +405,6 @@ func (u *UpstreamResolver) ParseAddresses() error {
405
return nil
406
}
407
554
-func allNew(upstreams []Upstream) []*Upstream {
555
- results := make([]*Upstream, len(upstreams))
556
- for i := range upstreams {
557
- results[i] = &Upstream{Dial: upstreams[i].Dial}
558
- }
559
- return results
560
-}
561
-
562
-var (
563
- srvs = make(map[string]srvLookup)
564
- srvsMu sync.RWMutex
565
-
566
- aAaaa = make(map[string]aLookup)
567
- aAaaaMu sync.RWMutex
568
-)
569
-
408
// Interface guards
409
var (
410
_ caddy.Provisioner = (*SRVUpstreams)(nil)
modules/internal/network/dynamicupstreams.go
new
+219
@@ -0,0 +1,219 @@
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 provides transport-neutral DNS-based discovery of upstream
16
+// targets, with result caching. It lives under modules/internal so it can be
17
+// shared between Caddy's own proxies (currently the HTTP reverse_proxy) without
18
+// each copying the resolution and caching logic.
19
+//
20
+// The package intentionally returns neutral [Target] values rather than any
21
+// proxy-specific upstream type; each caller builds its own upstream
22
+// representation from the targets.
23
+package network
24
+
25
+import (
26
+ "context"
27
+ "net"
28
+ "strconv"
29
+ "sync"
30
+ "time"
31
+
32
+ "go.uber.org/zap"
33
+ "go.uber.org/zap/zapcore"
34
+)
35
+
36
+// Target is a single discovered upstream endpoint. It carries the SRV priority
37
+// and weight when available (both zero for non-SRV lookups).
38
+type Target struct {
39
+ Host string
40
+ Port string
41
+ Priority uint16
42
+ Weight uint16
43
+}
44
+
45
+// SRVLookupFunc resolves SRV records. It matches the signature of
46
+// (*net.Resolver).LookupSRV, so callers can pass that directly (and tests can
47
+// inject a stub).
48
+type SRVLookupFunc func(ctx context.Context, service, proto, name string) (cname string, addrs []*net.SRV, err error)
49
+
50
+// SRV resolves the given SRV record into targets, caching the result for
51
+// refresh. lookup is the resolver to use (typically (*net.Resolver).LookupSRV).
52
+//
53
+// If the lookup fails but returns some records (e.g. a few invalid names were
54
+// filtered out), those records are still used. If it fails with no records and
55
+// grace > 0, the previously cached result keeps being served for up to grace
56
+// past its refresh instead of returning an error. With no cached result, grace
57
+// also suppresses the error and returns no targets, preserving the reverse
58
+// proxy's existing behavior.
59
+func SRV(ctx context.Context, lookup SRVLookupFunc, service, proto, name string, refresh, grace time.Duration, logger *zap.Logger) ([]Target, error) {
60
+ key := srvKey(service, proto, name)
61
+
62
+ // fast path: a fresh cached result under a read lock
63
+ srvMu.RLock()
64
+ cached := srvCache[key]
65
+ srvMu.RUnlock()
66
+ if cached.isFresh() {
67
+ return cached.targets, nil
68
+ }
69
+
70
+ srvMu.Lock()
71
+ defer srvMu.Unlock()
72
+
73
+ // re-check under the write lock in case another goroutine refreshed it
74
+ cached = srvCache[key]
75
+ if cached.isFresh() {
76
+ return cached.targets, nil
77
+ }
78
+
79
+ if c := logger.Check(zapcore.DebugLevel, "refreshing SRV upstreams"); c != nil {
80
+ c.Write(zap.String("service", service), zap.String("proto", proto), zap.String("name", name))
81
+ }
82
+
83
+ _, records, err := lookup(ctx, service, proto, name)
84
+ if err != nil {
85
+ // From LookupSRV docs: invalid names are filtered out and an error is
86
+ // returned alongside any remaining results; only treat it as fatal when
87
+ // nothing usable came back.
88
+ if len(records) == 0 {
89
+ if grace > 0 {
90
+ if c := logger.Check(zapcore.ErrorLevel, "SRV lookup failed; using previously cached"); c != nil {
91
+ c.Write(zap.Error(err))
92
+ }
93
+ cached.freshness = time.Now().Add(grace - refresh)
94
+ srvCache[key] = cached
95
+ return cached.targets, nil
96
+ }
97
+ return nil, err
98
+ }
99
+ if c := logger.Check(zapcore.WarnLevel, "SRV records filtered"); c != nil {
100
+ c.Write(zap.Error(err))
101
+ }
102
+ }
103
+
104
+ targets := make([]Target, len(records))
105
+ for i, rec := range records {
106
+ if c := logger.Check(zapcore.DebugLevel, "discovered SRV record"); c != nil {
107
+ c.Write(zap.String("target", rec.Target), zap.Uint16("port", rec.Port),
108
+ zap.Uint16("priority", rec.Priority), zap.Uint16("weight", rec.Weight))
109
+ }
110
+ targets[i] = Target{
111
+ Host: rec.Target,
112
+ Port: strconv.Itoa(int(rec.Port)),
113
+ Priority: rec.Priority,
114
+ Weight: rec.Weight,
115
+ }
116
+ }
117
+
118
+ // when inserting a brand-new entry (not replacing a stale one), bound the cache
119
+ if cached.freshness.IsZero() && len(srvCache) >= maxCacheEntries {
120
+ for k := range srvCache {
121
+ delete(srvCache, k)
122
+ break
123
+ }
124
+ }
125
+ srvCache[key] = cacheEntry{refresh: refresh, freshness: time.Now(), targets: targets}
126
+ return targets, nil
127
+}
128
+
129
+// ResetSRV removes the cached result for a single SRV record.
130
+func ResetSRV(service, proto, name string) {
131
+ srvMu.Lock()
132
+ delete(srvCache, srvKey(service, proto, name))
133
+ srvMu.Unlock()
134
+}
135
+
136
+// ResetAllSRV clears the entire SRV cache.
137
+func ResetAllSRV() {
138
+ srvMu.Lock()
139
+ srvCache = make(map[string]cacheEntry)
140
+ srvMu.Unlock()
141
+}
142
+
143
+func srvKey(service, proto, name string) string {
144
+ return service + "\x00" + proto + "\x00" + name
145
+}
146
+
147
+const maxCacheEntries = 100
148
+
149
+type cacheEntry struct {
150
+ refresh time.Duration
151
+ freshness time.Time
152
+ targets []Target
153
+}
154
+
155
+func (e cacheEntry) isFresh() bool {
156
+ return !e.freshness.IsZero() && time.Since(e.freshness) < e.refresh
157
+}
158
+
159
+// IPLookupFunc resolves a host's IP addresses. It matches the signature of
160
+// (*net.Resolver).LookupIP, so callers can pass that directly (and tests can
161
+// inject a stub). network is one of "ip", "ip4", "ip6".
162
+type IPLookupFunc func(ctx context.Context, network, host string) ([]net.IP, error)
163
+
164
+// A resolves name's A/AAAA records into targets (one per address, all using the
165
+// given port), caching the result for refresh. network selects the IP versions
166
+// ("ip", "ip4" or "ip6"). lookup is the resolver to use (typically
167
+// (*net.Resolver).LookupIP).
168
+func A(ctx context.Context, lookup IPLookupFunc, network, name, port string, refresh time.Duration, logger *zap.Logger) ([]Target, error) {
169
+ key := name + "\x00" + port + "\x00" + network
170
+
171
+ aMu.RLock()
172
+ cached := aCache[key]
173
+ aMu.RUnlock()
174
+ if cached.isFresh() {
175
+ return cached.targets, nil
176
+ }
177
+
178
+ aMu.Lock()
179
+ defer aMu.Unlock()
180
+
181
+ cached = aCache[key]
182
+ if cached.isFresh() {
183
+ return cached.targets, nil
184
+ }
185
+
186
+ if c := logger.Check(zapcore.DebugLevel, "refreshing A upstreams"); c != nil {
187
+ c.Write(zap.String("version", network), zap.String("name", name), zap.String("port", port))
188
+ }
189
+
190
+ ips, err := lookup(ctx, network, name)
191
+ if err != nil {
192
+ return nil, err
193
+ }
194
+
195
+ targets := make([]Target, len(ips))
196
+ for i, ip := range ips {
197
+ if c := logger.Check(zapcore.DebugLevel, "discovered A record"); c != nil {
198
+ c.Write(zap.String("ip", ip.String()))
199
+ }
200
+ targets[i] = Target{Host: ip.String(), Port: port}
201
+ }
202
+
203
+ if cached.freshness.IsZero() && len(aCache) >= maxCacheEntries {
204
+ for k := range aCache {
205
+ delete(aCache, k)
206
+ break
207
+ }
208
+ }
209
+ aCache[key] = cacheEntry{refresh: refresh, freshness: time.Now(), targets: targets}
210
+ return targets, nil
211
+}
212
+
213
+var (
214
+ srvMu sync.RWMutex
215
+ srvCache = make(map[string]cacheEntry)
216
+
217
+ aMu sync.RWMutex
218
+ aCache = make(map[string]cacheEntry)
219
+)
modules/internal/network/dynamicupstreams_test.go
new
+407
@@ -0,0 +1,407 @@
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
+ "fmt"
21
+ "net"
22
+ "sync"
23
+ "testing"
24
+ "time"
25
+
26
+ "go.uber.org/zap"
27
+ "go.uber.org/zap/zaptest"
28
+ "go.uber.org/zap/zaptest/observer"
29
+)
30
+
31
+func TestSRVResolvesAndCaches(t *testing.T) {
32
+ calls := 0
33
+ lookup := func(context.Context, string, string, string) (string, []*net.SRV, error) {
34
+ calls++
35
+ return "", []*net.SRV{
36
+ {Target: "a.example.", Port: 5432, Priority: 1, Weight: 10},
37
+ {Target: "b.example.", Port: 5433, Priority: 1, Weight: 20},
38
+ }, nil
39
+ }
40
+
41
+ targets, err := SRV(context.Background(), lookup, "svc-cache", "tcp", "x", time.Minute, 0, zap.NewNop())
42
+ if err != nil {
43
+ t.Fatalf("SRV: %v", err)
44
+ }
45
+ if len(targets) != 2 {
46
+ t.Fatalf("targets = %d, want 2", len(targets))
47
+ }
48
+ if targets[0].Host != "a.example." || targets[0].Port != "5432" || targets[0].Weight != 10 {
49
+ t.Errorf("unexpected first target: %+v", targets[0])
50
+ }
51
+
52
+ // second call within refresh must be served from cache (no extra lookup)
53
+ if _, err := SRV(context.Background(), lookup, "svc-cache", "tcp", "x", time.Minute, 0, zap.NewNop()); err != nil {
54
+ t.Fatalf("SRV (cached): %v", err)
55
+ }
56
+ if calls != 1 {
57
+ t.Errorf("lookup calls = %d, want 1 (cached)", calls)
58
+ }
59
+}
60
+
61
+func TestSRVErrorWithoutGrace(t *testing.T) {
62
+ lookup := func(context.Context, string, string, string) (string, []*net.SRV, error) {
63
+ return "", nil, errors.New("dns boom")
64
+ }
65
+ if _, err := SRV(context.Background(), lookup, "svc-err", "tcp", "x", time.Minute, 0, zap.NewNop()); err == nil {
66
+ t.Fatal("expected an error when lookup fails and nothing is cached")
67
+ }
68
+}
69
+
70
+func TestSRVGracePeriodServesStale(t *testing.T) {
71
+ ok := func(context.Context, string, string, string) (string, []*net.SRV, error) {
72
+ return "", []*net.SRV{{Target: "a.example.", Port: 1}}, nil
73
+ }
74
+ // tiny refresh so the entry is immediately stale on the next call
75
+ if _, err := SRV(context.Background(), ok, "svc-grace", "tcp", "x", time.Nanosecond, time.Hour, zap.NewNop()); err != nil {
76
+ t.Fatalf("seeding cache: %v", err)
77
+ }
78
+
79
+ fail := func(context.Context, string, string, string) (string, []*net.SRV, error) {
80
+ return "", nil, errors.New("dns boom")
81
+ }
82
+ // a level-enabled logger so the "using previously cached" error log fires
83
+ targets, err := SRV(context.Background(), fail, "svc-grace", "tcp", "x", time.Nanosecond, time.Hour, zaptest.NewLogger(t))
84
+ if err != nil {
85
+ t.Fatalf("grace period should suppress the error: %v", err)
86
+ }
87
+ if len(targets) != 1 {
88
+ t.Errorf("expected the stale cached target to be served, got %d", len(targets))
89
+ }
90
+}
91
+
92
+func TestResetSRV(t *testing.T) {
93
+ calls := 0
94
+ lookup := func(context.Context, string, string, string) (string, []*net.SRV, error) {
95
+ calls++
96
+ return "", []*net.SRV{{Target: "a.example.", Port: 1}}, nil
97
+ }
98
+ if _, err := SRV(context.Background(), lookup, "svc-reset", "tcp", "x", time.Minute, 0, zap.NewNop()); err != nil {
99
+ t.Fatal(err)
100
+ }
101
+ ResetSRV("svc-reset", "tcp", "x")
102
+ if _, err := SRV(context.Background(), lookup, "svc-reset", "tcp", "x", time.Minute, 0, zap.NewNop()); err != nil {
103
+ t.Fatal(err)
104
+ }
105
+ if calls != 2 {
106
+ t.Errorf("lookup calls = %d, want 2 (cache was reset between calls)", calls)
107
+ }
108
+}
109
+
110
+func TestAResolvesAndCaches(t *testing.T) {
111
+ calls := 0
112
+ lookup := func(context.Context, string, string) ([]net.IP, error) {
113
+ calls++
114
+ return []net.IP{net.ParseIP("10.0.0.1"), net.ParseIP("10.0.0.2")}, nil
115
+ }
116
+
117
+ targets, err := A(context.Background(), lookup, "ip", "db.a-test", "5432", time.Minute, zap.NewNop())
118
+ if err != nil {
119
+ t.Fatalf("A: %v", err)
120
+ }
121
+ if len(targets) != 2 {
122
+ t.Fatalf("targets = %d, want 2", len(targets))
123
+ }
124
+ if targets[0].Host != "10.0.0.1" || targets[0].Port != "5432" {
125
+ t.Errorf("unexpected first target: %+v", targets[0])
126
+ }
127
+
128
+ if _, err := A(context.Background(), lookup, "ip", "db.a-test", "5432", time.Minute, zap.NewNop()); err != nil {
129
+ t.Fatalf("A (cached): %v", err)
130
+ }
131
+ if calls != 1 {
132
+ t.Errorf("lookup calls = %d, want 1 (cached)", calls)
133
+ }
134
+}
135
+
136
+func TestAError(t *testing.T) {
137
+ lookup := func(context.Context, string, string) ([]net.IP, error) {
138
+ return nil, errors.New("dns boom")
139
+ }
140
+ if _, err := A(context.Background(), lookup, "ip", "db.a-err", "5432", time.Minute, zap.NewNop()); err == nil {
141
+ t.Fatal("expected an error when the A lookup fails")
142
+ }
143
+}
144
+
145
+// TestSRVFilteredRecords covers the LookupSRV semantics where invalid names are
146
+// filtered out and an error is returned alongside the usable remainder: the
147
+// usable records must still be returned (the error is downgraded to a warning).
148
+func TestSRVFilteredRecords(t *testing.T) {
149
+ lookup := func(context.Context, string, string, string) (string, []*net.SRV, error) {
150
+ return "", []*net.SRV{{Target: "ok.example.", Port: 5432, Priority: 1, Weight: 10}},
151
+ errors.New("some SRV names were filtered out")
152
+ }
153
+ // a level-enabled logger so the "SRV records filtered" warning fires
154
+ targets, err := SRV(context.Background(), lookup, "svc-filtered", "tcp", "x", time.Minute, 0, zaptest.NewLogger(t))
155
+ if err != nil {
156
+ t.Fatalf("usable records must be returned despite the partial error: %v", err)
157
+ }
158
+ if len(targets) != 1 || targets[0].Host != "ok.example." {
159
+ t.Fatalf("unexpected targets: %+v", targets)
160
+ }
161
+}
162
+
163
+// TestResetAllSRV verifies that the whole SRV cache is dropped.
164
+func TestResetAllSRV(t *testing.T) {
165
+ lookup := func(context.Context, string, string, string) (string, []*net.SRV, error) {
166
+ return "", []*net.SRV{{Target: "a.example.", Port: 1}}, nil
167
+ }
168
+ if _, err := SRV(context.Background(), lookup, "svc-reset-all", "tcp", "x", time.Minute, 0, zap.NewNop()); err != nil {
169
+ t.Fatalf("SRV: %v", err)
170
+ }
171
+ srvMu.RLock()
172
+ populated := len(srvCache) > 0
173
+ srvMu.RUnlock()
174
+ if !populated {
175
+ t.Fatal("expected the SRV cache to be populated before reset")
176
+ }
177
+
178
+ ResetAllSRV()
179
+
180
+ srvMu.RLock()
181
+ n := len(srvCache)
182
+ srvMu.RUnlock()
183
+ if n != 0 {
184
+ t.Fatalf("srvCache len = %d after ResetAllSRV, want 0", n)
185
+ }
186
+}
187
+
188
+// TestSRVCacheBound verifies that inserting a brand-new entry once the cache is
189
+// full evicts an existing one so the cache stays bounded.
190
+func TestSRVCacheBound(t *testing.T) {
191
+ srvMu.Lock()
192
+ srvCache = make(map[string]cacheEntry)
193
+ srvMu.Unlock()
194
+
195
+ lookup := func(context.Context, string, string, string) (string, []*net.SRV, error) {
196
+ return "", []*net.SRV{{Target: "a.example.", Port: 1}}, nil
197
+ }
198
+ for i := 0; i < maxCacheEntries+5; i++ {
199
+ name := fmt.Sprintf("svc-bound-%d", i)
200
+ if _, err := SRV(context.Background(), lookup, name, "tcp", "x", time.Minute, 0, zap.NewNop()); err != nil {
201
+ t.Fatalf("SRV[%d]: %v", i, err)
202
+ }
203
+ }
204
+ srvMu.RLock()
205
+ n := len(srvCache)
206
+ srvMu.RUnlock()
207
+ if n > maxCacheEntries {
208
+ t.Fatalf("srvCache len = %d, want <= %d (cache must stay bounded)", n, maxCacheEntries)
209
+ }
210
+}
211
+
212
+// TestACacheBound verifies the same bounding behavior for the A cache.
213
+func TestACacheBound(t *testing.T) {
214
+ aMu.Lock()
215
+ aCache = make(map[string]cacheEntry)
216
+ aMu.Unlock()
217
+
218
+ lookup := func(context.Context, string, string) ([]net.IP, error) {
219
+ return []net.IP{net.ParseIP("10.0.0.1")}, nil
220
+ }
221
+ for i := 0; i < maxCacheEntries+5; i++ {
222
+ name := fmt.Sprintf("db.bound-%d", i)
223
+ if _, err := A(context.Background(), lookup, "ip", name, "5432", time.Minute, zap.NewNop()); err != nil {
224
+ t.Fatalf("A[%d]: %v", i, err)
225
+ }
226
+ }
227
+ aMu.RLock()
228
+ n := len(aCache)
229
+ aMu.RUnlock()
230
+ if n > maxCacheEntries {
231
+ t.Fatalf("aCache len = %d, want <= %d (cache must stay bounded)", n, maxCacheEntries)
232
+ }
233
+}
234
+
235
+// TestSRVConcurrentRefreshDeduplicates verifies that two concurrent callers
236
+// missing the cache for the same key result in a single lookup: the first
237
+// performs it while holding the write lock, and the second is served the
238
+// cached result once the first releases. Depending on scheduling the second
239
+// caller may be served by either the read-lock fast path or the write-lock
240
+// re-check — this test does not force one branch over the other; it asserts
241
+// deduplication (exactly one lookup). Run under -race it also guards against
242
+// data races on the shared cache.
243
+func TestSRVConcurrentRefreshDeduplicates(t *testing.T) {
244
+ srvMu.Lock()
245
+ srvCache = make(map[string]cacheEntry)
246
+ srvMu.Unlock()
247
+
248
+ var calls int
249
+ var mu sync.Mutex
250
+ inLookup := make(chan struct{})
251
+ release := make(chan struct{})
252
+ lookup := func(context.Context, string, string, string) (string, []*net.SRV, error) {
253
+ mu.Lock()
254
+ calls++
255
+ first := calls == 1
256
+ mu.Unlock()
257
+ if first {
258
+ close(inLookup)
259
+ <-release // hold the write lock until the second goroutine is queued
260
+ }
261
+ return "", []*net.SRV{{Target: "a.example.", Port: 1}}, nil
262
+ }
263
+
264
+ done := make(chan struct{}, 2)
265
+ // G1: takes the write lock and blocks inside lookup.
266
+ go func() {
267
+ _, _ = SRV(context.Background(), lookup, "svc-conc", "tcp", "x", time.Minute, 0, zap.NewNop())
268
+ done <- struct{}{}
269
+ }()
270
+ <-inLookup // G1 now holds the write lock; cache is still empty
271
+
272
+ // G2: a second caller for the same key. While G1 holds the write lock, G2
273
+ // cannot acquire srvMu at all (a reader blocks behind the writer); once G1
274
+ // populates the cache and releases, G2 observes the cached entry and
275
+ // performs no lookup. Ordering here is enforced by the channels below, not
276
+ // by timing.
277
+ go func() {
278
+ _, _ = SRV(context.Background(), lookup, "svc-conc", "tcp", "x", time.Minute, 0, zap.NewNop())
279
+ done <- struct{}{}
280
+ }()
281
+
282
+ close(release) // G1 populates the cache and releases the lock
283
+ <-done
284
+ <-done
285
+
286
+ mu.Lock()
287
+ got := calls
288
+ mu.Unlock()
289
+ if got != 1 {
290
+ t.Fatalf("lookup calls = %d, want 1 (concurrent callers for the same key must share one lookup)", got)
291
+ }
292
+}
293
+
294
+// TestAConcurrentRefreshDeduplicates is the A-cache equivalent of the above:
295
+// it asserts deduplication (one lookup for two concurrent callers) and, under
296
+// -race, race-safety; it does not force a particular locking branch.
297
+func TestAConcurrentRefreshDeduplicates(t *testing.T) {
298
+ aMu.Lock()
299
+ aCache = make(map[string]cacheEntry)
300
+ aMu.Unlock()
301
+
302
+ var calls int
303
+ var mu sync.Mutex
304
+ inLookup := make(chan struct{})
305
+ release := make(chan struct{})
306
+ lookup := func(context.Context, string, string) ([]net.IP, error) {
307
+ mu.Lock()
308
+ calls++
309
+ first := calls == 1
310
+ mu.Unlock()
311
+ if first {
312
+ close(inLookup)
313
+ <-release
314
+ }
315
+ return []net.IP{net.ParseIP("10.0.0.1")}, nil
316
+ }
317
+
318
+ done := make(chan struct{}, 2)
319
+ go func() {
320
+ _, _ = A(context.Background(), lookup, "ip", "db.conc", "5432", time.Minute, zap.NewNop())
321
+ done <- struct{}{}
322
+ }()
323
+ <-inLookup
324
+
325
+ // Second caller for the same key; ordering is enforced by the channels,
326
+ // not by timing (see the SRV test above).
327
+ go func() {
328
+ _, _ = A(context.Background(), lookup, "ip", "db.conc", "5432", time.Minute, zap.NewNop())
329
+ done <- struct{}{}
330
+ }()
331
+
332
+ close(release)
333
+ <-done
334
+ <-done
335
+
336
+ mu.Lock()
337
+ got := calls
338
+ mu.Unlock()
339
+ if got != 1 {
340
+ t.Fatalf("lookup calls = %d, want 1 (concurrent callers for the same key must share one lookup)", got)
341
+ }
342
+}
343
+
344
+func TestSRVGracePeriodWithoutCachedResult(t *testing.T) {
345
+ ResetSRV("svc-cold-grace", "tcp", "x")
346
+ t.Cleanup(func() { ResetSRV("svc-cold-grace", "tcp", "x") })
347
+ lookup := func(context.Context, string, string, string) (string, []*net.SRV, error) {
348
+ return "", nil, errors.New("dns boom")
349
+ }
350
+ targets, err := SRV(context.Background(), lookup, "svc-cold-grace", "tcp", "x", time.Minute, time.Minute, zap.NewNop())
351
+ if err != nil || len(targets) != 0 {
352
+ t.Fatalf("cold-cache failure with grace: targets = %v, err = %v; want no targets and no error", targets, err)
353
+ }
354
+}
355
+
356
+func TestDiscoveryLoggingOnlyOnRefresh(t *testing.T) {
357
+ t.Run("SRV", func(t *testing.T) {
358
+ ResetSRV("svc-log", "tcp", "x")
359
+ t.Cleanup(func() { ResetSRV("svc-log", "tcp", "x") })
360
+ core, logs := observer.New(zap.DebugLevel)
361
+ logger := zap.New(core)
362
+ lookup := func(context.Context, string, string, string) (string, []*net.SRV, error) {
363
+ return "", []*net.SRV{{Target: "a.example.", Port: 5432, Priority: 1, Weight: 10}}, nil
364
+ }
365
+ for range 2 {
366
+ if _, err := SRV(context.Background(), lookup, "svc-log", "tcp", "x", time.Minute, 0, logger); err != nil {
367
+ t.Fatal(err)
368
+ }
369
+ }
370
+ if got := logs.FilterMessage("refreshing SRV upstreams").Len(); got != 1 {
371
+ t.Fatalf("refresh logs = %d, want 1", got)
372
+ }
373
+ records := logs.FilterMessage("discovered SRV record").All()
374
+ if len(records) != 1 {
375
+ t.Fatalf("discovery logs = %d, want 1", len(records))
376
+ }
377
+ fields := records[0].ContextMap()
378
+ if fields["target"] != "a.example." || fields["port"] != uint16(5432) || fields["priority"] != uint16(1) || fields["weight"] != uint16(10) {
379
+ t.Fatalf("unexpected discovery log fields: %v", fields)
380
+ }
381
+ })
382
+ t.Run("A", func(t *testing.T) {
383
+ aMu.Lock()
384
+ aCache = make(map[string]cacheEntry)
385
+ aMu.Unlock()
386
+ core, logs := observer.New(zap.DebugLevel)
387
+ logger := zap.New(core)
388
+ lookup := func(context.Context, string, string) ([]net.IP, error) {
389
+ return []net.IP{net.ParseIP("10.0.0.1")}, nil
390
+ }
391
+ for range 2 {
392
+ if _, err := A(context.Background(), lookup, "ip", "db.log", "5432", time.Minute, logger); err != nil {
393
+ t.Fatal(err)
394
+ }
395
+ }
396
+ if got := logs.FilterMessage("refreshing A upstreams").Len(); got != 1 {
397
+ t.Fatalf("refresh logs = %d, want 1", got)
398
+ }
399
+ records := logs.FilterMessage("discovered A record").All()
400
+ if len(records) != 1 {
401
+ t.Fatalf("discovery logs = %d, want 1", len(records))
402
+ }
403
+ if got := records[0].ContextMap()["ip"]; got != "10.0.0.1" {
404
+ t.Fatalf("logged IP = %v, want 10.0.0.1", got)
405
+ }
406
+ })
407
+}