diff --git a/changelog/v0.41.2/retain-unanswered-xds-watches.yaml b/changelog/v0.41.2/retain-unanswered-xds-watches.yaml new file mode 100644 index 000000000..01c4d6423 --- /dev/null +++ b/changelog/v0.41.2/retain-unanswered-xds-watches.yaml @@ -0,0 +1,35 @@ +changelog: + - type: FIX + issueLink: https://github.com/solo-io/solo-kit/pull/586 + resolvesIssue: false + description: >- + Keep an xDS watch open when its response has to be withheld. In ADS mode + the snapshot cache does not answer a request whose resource names do not + cover the snapshot's resources of that type, which happens while a client + has not yet subscribed to a resource the snapshot already carries. Both + call sites treated the withheld response as a completed request, so the + watch was discarded on a snapshot update and never registered for a new + request. Because a client only issues a new request after receiving a + response, nothing re-armed delivery, and it stayed on its previous + configuration until it reconnected. When enabled, the watch is retained on both + paths and answered by the first snapshot it can accept. + Set EXP_FIX to "off" to restore the previous + behavior on both paths, or "on" to force retention. "default", unset, and + empty values follow the release default (currently off), which may change + in a future release; explicit "on" and "off" remain pinned. Other values + log a warning and use the release default. The setting is read when each + snapshot cache is created. + - type: NEW_FEATURE + issueLink: https://github.com/solo-io/solo-kit/pull/587 + resolvesIssue: false + description: >- + Add the xds/suppressed_responses measure, counting the times the snapshot + cache intentionally withholds an xDS response, tagged with the resource + type and a reason. The only reason today is ads_name_mismatch: the + request's resource names do not cover the snapshot's resources of that + type. The measure is recorded whether or not watch retention is enabled, + because nothing else reports that the condition happened -- with retention + off the watch is silently discarded, and with it on the watch survives but + still goes unanswered until a compatible snapshot arrives. It is therefore + the signal for how often this occurs in a running fleet, and the basis for + deciding the release default of EXP_FIX. diff --git a/pkg/api/v1/control-plane/cache/simple.go b/pkg/api/v1/control-plane/cache/simple.go index 529d62905..154fb811c 100644 --- a/pkg/api/v1/control-plane/cache/simple.go +++ b/pkg/api/v1/control-plane/cache/simple.go @@ -18,9 +18,13 @@ import ( "context" "errors" "fmt" + "os" + "sync" "time" + "github.com/solo-io/go-utils/contextutils" + "github.com/solo-io/solo-kit/pkg/api/v1/control-plane/log" "go.opencensus.io/stats" @@ -29,9 +33,11 @@ import ( ) var ( - MResponses = stats.Int64("xds/responses", "The responses for envoy", "1") + MResponses = stats.Int64("xds/responses", "The responses for envoy", "1") + MSuppressedResponses = stats.Int64("xds/suppressed_responses", "The responses intentionally suppressed by cache logic", "1") - KeyType = tag.MustNewKey("type") + KeyType = tag.MustNewKey("type") + KeyReason = tag.MustNewKey("reason") ResponsesView = &view.View{ Name: "xds/responses", @@ -43,14 +49,35 @@ var ( }, } + // SuppressedResponsesView counts responses the cache withholds. It is + // recorded whether or not watch retention is enabled, because it is the only + // signal that says how often the condition fires at all: with retention off + // a withheld response silently discards the watch, and with it on the watch + // is kept but still goes unanswered until a compatible snapshot arrives. + // Either way nothing else in the cache reports that it happened. + SuppressedResponsesView = &view.View{ + Name: "xds/suppressed_responses", + Measure: MSuppressedResponses, + Description: "The times the cache intentionally withheld an xDS response", + Aggregation: view.Count(), + TagKeys: []tag.Key{ + KeyType, + KeyReason, + }, + } + VersionUpToDateError = errors.New("skip fetch: version up to date") // Compile-time assertion _ SnapshotCache = new(snapshotCache) ) +// suppressedResponseReasonAdsNameMismatch is recorded when ADS withholds a +// response because the request does not name every resource in the snapshot. +const suppressedResponseReasonAdsNameMismatch = "ads_name_mismatch" + func init() { - view.Register(ResponsesView) + view.Register(ResponsesView, SuppressedResponsesView) } // SnapshotCache is a snapshot-based cache that maintains a single versioned @@ -100,8 +127,38 @@ type CacheSettings struct { PrioritySet map[int][]string } +// RetainUnansweredXDSWatchesEnv controls retention of withheld ADS watches. +// "on" and "off" pin the behavior; "default", unset, and empty follow the +// release default. Invalid values warn and also follow the release default. +const RetainUnansweredXDSWatchesEnv = "EXP_FIX" + +// Change this default independently of the explicit on/off overrides. +const retainUnansweredXDSWatchesDefault = false + +func retainUnansweredXDSWatches(logger log.Logger) bool { + value := os.Getenv(RetainUnansweredXDSWatchesEnv) + switch value { + case "on": + return true + case "off": + return false + case "", "default": + return retainUnansweredXDSWatchesDefault + default: + // The cache logger predates warning support. Use it when available, + // otherwise use the standard application logger without changing its interface. + warningLogger, ok := logger.(interface{ Warnf(string, ...interface{}) }) + if !ok { + warningLogger = contextutils.LoggerFrom(context.Background()) + } + warningLogger.Warnf("invalid %s value %q: expected default, on, or off; using default (%t)", RetainUnansweredXDSWatchesEnv, value, retainUnansweredXDSWatchesDefault) + return retainUnansweredXDSWatchesDefault + } +} + type snapshotCache struct { - log log.Logger + retainUnansweredWatches bool + log log.Logger // ads flag to hold responses until all resources are named ads bool @@ -132,14 +189,16 @@ type snapshotCache struct { // is OK. // // Logger is optional. +// RetainUnansweredXDSWatchesEnv is read once when each cache is created. func NewSnapshotCache(settings CacheSettings) SnapshotCache { return &snapshotCache{ - log: settings.Logger, - ads: settings.Ads, - snapshots: make(map[string]Snapshot), - status: make(map[string]*statusInfo), - hash: settings.Hash, - prioritySet: settings.PrioritySet, + retainUnansweredWatches: retainUnansweredXDSWatches(settings.Logger), + log: settings.Logger, + ads: settings.Ads, + snapshots: make(map[string]Snapshot), + status: make(map[string]*statusInfo), + hash: settings.Hash, + prioritySet: settings.PrioritySet, } } @@ -168,9 +227,17 @@ func (cache *snapshotCache) SetSnapshot(node string, snapshot Snapshot) { // empty resource - intended behavior is for the resources to be cleared if resources != nil { // snapshot has been initialized and exists - cache.respond(watch.Request, watch.Response, resources, version) - - // discard the watch + if sent := cache.respond(watch.Request, watch.Response, resources, version); !sent && cache.retainUnansweredWatches { + // The response was withheld (see respond): keep the watch so a + // later snapshot that this request can accept is delivered to it. + if cache.log != nil { + cache.log.Debugf("keeping open watch priority %d and index %d :%v; version %q could not be sent", + pi.Index, pi.Priority, watch.Request.ResourceNames, version) + } + return + } + + // Discard answered watches, or all watches when retention is disabled. watches.Delete(pi) } } @@ -266,25 +333,32 @@ func (cache *snapshotCache) CreateWatch(request Request) (chan Response, func()) } version := snapshot.GetResources(request.TypeUrl).Version - // if the requested version is up-to-date or missing a response, leave an open watch - if !exists || request.VersionInfo == version { - info.mu.Lock() - // check SetSnapshot() for responses on the watches map - priority := info.watches.Add(ResponseWatch{Request: request, Response: value}) - info.mu.Unlock() - if cache.log != nil { - cache.log.Debugf("open watch Priority Index %d and Element Index %d for %s%v from nodeID %q, version %q", - priority.Priority, priority.Index, request.TypeUrl, request.ResourceNames, nodeID, request.VersionInfo) + // If the snapshot is newer than the requested version, try to answer the + // request immediately. The response can still be withheld in ADS mode (see + // respond), in which case retention opens a watch: the + // client is waiting for this response and does not issue a new request + // until it receives one, so dropping the request here would leave the + // client on its current configuration until it reconnects. + if exists && request.VersionInfo != version { + if sent := cache.respond(request, value, snapshot.GetResources(request.TypeUrl).Items, version); sent || !cache.retainUnansweredWatches { + return value, func() { + close(value) + } } - return value, cache.cancelWatch(nodeID, priority) } - // otherwise, the watch may be responded immediately - cache.respond(request, value, snapshot.GetResources(request.TypeUrl).Items, version) - - return value, func() { - close(value) + // Leave an open watch: the requested version is up to date, there is no + // snapshot for this node yet, or the response above was withheld. + info.mu.Lock() + // check SetSnapshot() for responses on the watches map + priority := info.watches.Add(ResponseWatch{Request: request, Response: value}) + info.mu.Unlock() + if cache.log != nil { + cache.log.Debugf("open watch Priority Index %d and Element Index %d for %s%v from nodeID %q, version %q", + priority.Priority, priority.Index, request.TypeUrl, request.ResourceNames, nodeID, request.VersionInfo) } + + return value, cache.cancelWatch(nodeID, priority) } // cancellation function for cleaning stale watches @@ -305,16 +379,27 @@ func (cache *snapshotCache) cancelWatch(nodeID string, watchIndex PriorityIndex) } // Respond to a watch with the snapshot value. The value channel should have capacity not to block. +// +// It reports whether a response was sent. In ADS mode a request whose resource +// names do not cover the snapshot cannot be answered yet, and callers must keep +// the watch open when retention is enabled: nothing else will +// answer it, because the client is waiting for this response and only issues a +// new request once it receives one. +// // TODO(kuat) do not respond always, see issue https://github.com/envoyproxy/go-control-plane/issues/46 -func (cache *snapshotCache) respond(request Request, value chan Response, resources map[string]Resource, version string) { +func (cache *snapshotCache) respond(request Request, value chan Response, resources map[string]Resource, version string) bool { // for ADS, the request names must match the snapshot names - // if they do not, then the watch is never responded, and it is expected that envoy makes another request + // if they do not, return false so callers can retain the watch when enabled if len(request.ResourceNames) != 0 && cache.ads { if err := Superset(nameSet(request.ResourceNames), resources); err != nil { if cache.log != nil { cache.log.Debugf("ADS mode: not responding to request: %v", err) } - return + stats.RecordWithTags(context.TODO(), []tag.Mutator{ + tag.Insert(KeyType, request.GetTypeUrl()), + tag.Insert(KeyReason, suppressedResponseReasonAdsNameMismatch), + }, MSuppressedResponses.M(1)) + return false } } if cache.log != nil { @@ -325,6 +410,8 @@ func (cache *snapshotCache) respond(request Request, value chan Response, resour tag.Insert(KeyType, request.GetTypeUrl()), }, MResponses.M(1)) value <- createResponse(request, resources, version) + + return true } func createResponse(request Request, resources map[string]Resource, version string) Response { diff --git a/pkg/api/v1/control-plane/cache/suppressed_responses_test.go b/pkg/api/v1/control-plane/cache/suppressed_responses_test.go new file mode 100644 index 000000000..437d9c992 --- /dev/null +++ b/pkg/api/v1/control-plane/cache/suppressed_responses_test.go @@ -0,0 +1,117 @@ +package cache_test + +import ( + envoy_config_core_v3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/solo-io/solo-kit/pkg/api/v1/control-plane/cache" + "github.com/solo-io/solo-kit/pkg/api/v1/control-plane/resource" + "github.com/solo-io/solo-kit/pkg/api/v1/control-plane/types" + "go.opencensus.io/stats/view" +) + +// A withheld response is otherwise invisible. With retention off the watch is +// silently discarded; with it on the watch survives but still goes unanswered +// until a compatible snapshot arrives. Neither case logs above debug or moves +// any other counter, so this measure is the only thing that says how often the +// condition fires -- which is what makes the retention default decidable from +// field data rather than from argument. +var _ = Describe("Suppressed response metric", func() { + + const suppressedNode = "suppressed-response-node" + + // suppressedCount totals the rows of xds/suppressed_responses. The view is + // process-wide and other specs record into it, so a test reads the delta + // across the action rather than an absolute. + suppressedCount := func() int64 { + rows, err := view.RetrieveData(cache.SuppressedResponsesView.Name) + Expect(err).NotTo(HaveOccurred()) + var total int64 + for _, row := range rows { + if data, ok := row.Data.(*view.CountData); ok { + total += data.Value + } + } + return total + } + + // withholdOneResponse asks for "a" while the snapshot also holds "b", which + // is the ADS name mismatch that makes the cache withhold. + withholdOneResponse := func(snapshotCache cache.SnapshotCache) func() { + snapshotCache.SetSnapshot(suppressedNode, &TestSnapshot{ + Endpoints: cache.NewResources("v1", []cache.Resource{ + resource.NewEnvoyResource(makeEndpoint("a")), + resource.NewEnvoyResource(makeEndpoint("b")), + }), + }) + _, cancel := snapshotCache.CreateWatch(cache.Request{ + TypeUrl: types.EndpointTypeV3, + ResourceNames: []string{"a"}, + VersionInfo: "v0", + Node: &envoy_config_core_v3.Node{Id: suppressedNode}, + }) + return cancel + } + + DescribeTable("counts a withheld response whether or not watches are retained", + func(retention string) { + GinkgoT().Setenv(cache.RetainUnansweredXDSWatchesEnv, retention) + snapshotCache := cache.NewSnapshotCache(cache.CacheSettings{Ads: true, Hash: TestIDHash{}}) + + before := suppressedCount() + cancel := withholdOneResponse(snapshotCache) + defer cancel() + + Expect(suppressedCount()).To(BeNumerically(">", before), + "a withheld response must be counted regardless of the retention setting") + }, + Entry("retention on", "on"), + Entry("retention off", "off"), + ) + + It("does not count a response that was sent", func() { + GinkgoT().Setenv(cache.RetainUnansweredXDSWatchesEnv, "on") + snapshotCache := cache.NewSnapshotCache(cache.CacheSettings{Ads: true, Hash: TestIDHash{}}) + snapshotCache.SetSnapshot(suppressedNode, &TestSnapshot{ + Endpoints: cache.NewResources("v1", []cache.Resource{ + resource.NewEnvoyResource(makeEndpoint("a")), + }), + }) + + before := suppressedCount() + // The request now names every resource in the snapshot, so it is answered. + _, cancel := snapshotCache.CreateWatch(cache.Request{ + TypeUrl: types.EndpointTypeV3, + ResourceNames: []string{"a"}, + VersionInfo: "v0", + Node: &envoy_config_core_v3.Node{Id: suppressedNode}, + }) + defer cancel() + + Expect(suppressedCount()).To(Equal(before)) + }) + + It("tags the reason so the cause is distinguishable", func() { + GinkgoT().Setenv(cache.RetainUnansweredXDSWatchesEnv, "on") + snapshotCache := cache.NewSnapshotCache(cache.CacheSettings{Ads: true, Hash: TestIDHash{}}) + cancel := withholdOneResponse(snapshotCache) + defer cancel() + + rows, err := view.RetrieveData(cache.SuppressedResponsesView.Name) + Expect(err).NotTo(HaveOccurred()) + + var reasons, typeURLs []string + for _, row := range rows { + for _, t := range row.Tags { + switch t.Key.Name() { + case "reason": + reasons = append(reasons, t.Value) + case "type": + typeURLs = append(typeURLs, t.Value) + } + } + } + Expect(reasons).To(ContainElement("ads_name_mismatch")) + Expect(typeURLs).To(ContainElement(types.EndpointTypeV3)) + }) +}) diff --git a/pkg/api/v1/control-plane/cache/watch_retention_env_test.go b/pkg/api/v1/control-plane/cache/watch_retention_env_test.go new file mode 100644 index 000000000..7be20b3f1 --- /dev/null +++ b/pkg/api/v1/control-plane/cache/watch_retention_env_test.go @@ -0,0 +1,82 @@ +package cache + +import ( + "fmt" + "os" + "strings" + "testing" +) + +type warningRecorder struct{ warnings []string } + +func (*warningRecorder) Debugf(string, ...interface{}) {} +func (*warningRecorder) Infof(string, ...interface{}) {} +func (*warningRecorder) Errorf(string, ...interface{}) {} +func (l *warningRecorder) Warnf(format string, args ...interface{}) { + l.warnings = append(l.warnings, fmt.Sprintf(format, args...)) +} + +func TestWatchRetentionEnvironment(t *testing.T) { + for _, tc := range []struct { + value string + enabled bool + warning bool + }{ + {"", retainUnansweredXDSWatchesDefault, false}, + {"default", retainUnansweredXDSWatchesDefault, false}, + {"on", true, false}, + {"off", false, false}, + {"invalid", retainUnansweredXDSWatchesDefault, true}, + {"ON", retainUnansweredXDSWatchesDefault, true}, + {" ", retainUnansweredXDSWatchesDefault, true}, + } { + t.Run(fmt.Sprintf("value=%q", tc.value), func(t *testing.T) { + t.Setenv(RetainUnansweredXDSWatchesEnv, tc.value) + logger := &warningRecorder{} + c := NewSnapshotCache(CacheSettings{Logger: logger}).(*snapshotCache) + if c.retainUnansweredWatches != tc.enabled { + t.Fatalf("retention = %t, want %t", c.retainUnansweredWatches, tc.enabled) + } + wantWarnings := 0 + if tc.warning { + wantWarnings = 1 + } + if len(logger.warnings) != wantWarnings { + t.Fatalf("warnings = %v, want %d", logger.warnings, wantWarnings) + } + if tc.warning && (!strings.Contains(logger.warnings[0], RetainUnansweredXDSWatchesEnv) || !strings.Contains(logger.warnings[0], fmt.Sprintf("%q", tc.value))) { + t.Fatalf("warning does not identify invalid setting: %s", logger.warnings[0]) + } + }) + } +} + +func TestWatchRetentionEnvironmentIsReadAtConstruction(t *testing.T) { + t.Setenv(RetainUnansweredXDSWatchesEnv, "on") + enabled := NewSnapshotCache(CacheSettings{}).(*snapshotCache) + t.Setenv(RetainUnansweredXDSWatchesEnv, "off") + disabled := NewSnapshotCache(CacheSettings{}).(*snapshotCache) + if !enabled.retainUnansweredWatches || disabled.retainUnansweredWatches { + t.Fatal("each cache must keep the setting read at construction") + } +} + +func TestWatchRetentionInvalidEnvironmentWithoutLogger(t *testing.T) { + t.Setenv(RetainUnansweredXDSWatchesEnv, "invalid") + c := NewSnapshotCache(CacheSettings{}).(*snapshotCache) + if c.retainUnansweredWatches != retainUnansweredXDSWatchesDefault { + t.Fatal("invalid setting must use the release default") + } +} + +func TestWatchRetentionUnsetEnvironment(t *testing.T) { + t.Setenv(RetainUnansweredXDSWatchesEnv, "") + if err := os.Unsetenv(RetainUnansweredXDSWatchesEnv); err != nil { + t.Fatal(err) + } + logger := &warningRecorder{} + c := NewSnapshotCache(CacheSettings{Logger: logger}).(*snapshotCache) + if c.retainUnansweredWatches != retainUnansweredXDSWatchesDefault || len(logger.warnings) != 0 { + t.Fatal("unset setting must use the release default without warning") + } +} diff --git a/pkg/api/v1/control-plane/cache/watch_retention_test.go b/pkg/api/v1/control-plane/cache/watch_retention_test.go new file mode 100644 index 000000000..f713a2f62 --- /dev/null +++ b/pkg/api/v1/control-plane/cache/watch_retention_test.go @@ -0,0 +1,161 @@ +package cache_test + +import ( + "time" + + envoy_config_core_v3 "github.com/envoyproxy/go-control-plane/envoy/config/core/v3" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" + "github.com/solo-io/solo-kit/pkg/api/v1/control-plane/cache" + "github.com/solo-io/solo-kit/pkg/api/v1/control-plane/resource" + "github.com/solo-io/solo-kit/pkg/api/v1/control-plane/types" +) + +// In ADS mode a response is withheld when the request's resource names do not +// cover the snapshot's resources of that type, which happens whenever the +// snapshot knows about a resource the client has not subscribed to yet (a +// cluster it has not applied, or one it rejected). The request must stay an +// open watch through that window: the client is waiting for a response and +// only issues a new request once it gets one, so a dropped watch leaves it on +// its current configuration until it reconnects. +var _ = Describe("Snapshot cache watch retention", func() { + const retentionNode = "watch-retention-node" + + var snapshotCache cache.SnapshotCache + + BeforeEach(func() { + GinkgoT().Setenv(cache.RetainUnansweredXDSWatchesEnv, "on") + snapshotCache = cache.NewSnapshotCache(cache.CacheSettings{Ads: true, Hash: TestIDHash{}}) + }) + + // endpointSnapshot builds a snapshot whose EDS resources are assignments + // for the named clusters. + endpointSnapshot := func(version string, clusters ...string) cache.Snapshot { + resources := make([]cache.Resource, 0, len(clusters)) + for _, name := range clusters { + resources = append(resources, resource.NewEnvoyResource(makeEndpoint(name))) + } + return cache.NewEasyGenericSnapshot(version, resources) + } + + endpointRequest := func(version string, names ...string) cache.Request { + return cache.Request{ + Node: &envoy_config_core_v3.Node{Id: retentionNode}, + TypeUrl: types.EndpointTypeV3, + ResourceNames: names, + VersionInfo: version, + } + } + + openWatches := func() int { + return snapshotCache.GetStatusInfo(retentionNode).GetNumWatches() + } + + It("keeps an open watch whose response a new snapshot cannot carry, and answers it once one can", func() { + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("1", "cluster-a")) + + initial, _ := snapshotCache.CreateWatch(endpointRequest("", "cluster-a")) + Eventually(initial, time.Second).Should(Receive()) + + // The client acks and re-requests: the watch is parked. + parked, cancel := snapshotCache.CreateWatch(endpointRequest("1", "cluster-a")) + defer cancel() + Expect(openWatches()).To(Equal(1)) + + // A snapshot that also carries an assignment for a cluster this client + // has not subscribed to cannot be sent to that watch. + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("2", "cluster-a", "cluster-b")) + Consistently(parked, 200*time.Millisecond, 20*time.Millisecond).ShouldNot(Receive()) + Expect(openWatches()).To(Equal(1), "the watch must survive a snapshot it cannot be sent") + + // Once the snapshot is one this request can accept, the retained watch + // receives it. + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("3", "cluster-a")) + var response cache.Response + Eventually(parked, time.Second).Should(Receive(&response)) + Expect(response.Version).To(Equal("3")) + Expect(openWatches()).To(Equal(0), "an answered watch is discarded") + }) + + It("opens a watch for a request the current snapshot cannot answer, and answers it once one can", func() { + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("2", "cluster-a", "cluster-b")) + + // The request asks for a newer version than it holds, so the cache + // tries to answer it immediately -- and cannot, because the snapshot + // carries cluster-b as well. + pending, cancel := snapshotCache.CreateWatch(endpointRequest("1", "cluster-a")) + defer cancel() + Consistently(pending, 200*time.Millisecond, 20*time.Millisecond).ShouldNot(Receive()) + Expect(openWatches()).To(Equal(1), "a request that could not be answered must become an open watch") + + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("3", "cluster-a")) + var response cache.Response + Eventually(pending, time.Second).Should(Receive(&response)) + Expect(response.Version).To(Equal("3")) + }) + + It("still answers a request the snapshot can carry immediately", func() { + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("2", "cluster-a")) + + answered, _ := snapshotCache.CreateWatch(endpointRequest("1", "cluster-a")) + var response cache.Response + Eventually(answered, time.Second).Should(Receive(&response)) + Expect(response.Version).To(Equal("2")) + Expect(openWatches()).To(Equal(0), "an immediately answered request leaves no watch behind") + }) + + It("still parks a request that is already up to date", func() { + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("1", "cluster-a")) + + parked, cancel := snapshotCache.CreateWatch(endpointRequest("1", "cluster-a")) + defer cancel() + Consistently(parked, 200*time.Millisecond, 20*time.Millisecond).ShouldNot(Receive()) + Expect(openWatches()).To(Equal(1)) + + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("2", "cluster-a")) + Eventually(parked, time.Second).Should(Receive()) + }) + + It("is unaffected for wildcard requests, which are never withheld", func() { + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("1", "cluster-a")) + + initial, _ := snapshotCache.CreateWatch(endpointRequest("")) + Eventually(initial, time.Second).Should(Receive()) + + parked, cancel := snapshotCache.CreateWatch(endpointRequest("1")) + defer cancel() + Expect(openWatches()).To(Equal(1)) + + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("2", "cluster-a", "cluster-b")) + Eventually(parked, time.Second).Should(Receive()) + Expect(openWatches()).To(Equal(0)) + }) + Context("with retention disabled", func() { + BeforeEach(func() { + GinkgoT().Setenv(cache.RetainUnansweredXDSWatchesEnv, "off") + snapshotCache = cache.NewSnapshotCache(cache.CacheSettings{Ads: true, Hash: TestIDHash{}}) + }) + + It("discards a parked watch even when the snapshot response is withheld", func() { + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("1", "cluster-a")) + pending, cancel := snapshotCache.CreateWatch(endpointRequest("1", "cluster-a")) + defer cancel() + Expect(openWatches()).To(Equal(1)) + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("2", "cluster-a", "cluster-b")) + Expect(openWatches()).To(Equal(0)) + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("3", "cluster-a")) + Expect(pending).ShouldNot(Receive()) + }) + + It("does not register a stale request whose immediate response is withheld", func() { + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("2", "cluster-a", "cluster-b")) + pending, cancel := snapshotCache.CreateWatch(endpointRequest("1", "cluster-a")) + Expect(openWatches()).To(Equal(0)) + snapshotCache.SetSnapshot(retentionNode, endpointSnapshot("3", "cluster-a")) + Expect(pending).ShouldNot(Receive()) + cancel() + Expect(pending).To(BeClosed()) + }) + }) + +})