incremental: don't delete non-existent resources in non-wildcard mode…

… in delta xds (#488)

* Respond with empty resource if it was explicitly requested by the client and doesn't exist on the server in delta xds

Signed-off-by: Sergey Matyukevich <s.matyukevich@gmail.com>

pkg/cache/v3/delta.go

@@ -56,13 +56,18 @@ func createDeltaResponse(ctx context.Context, req *DeltaRequest, state stream.St
 					filtered = append(filtered, r)
 				}
 				nextVersionMap[name] = nextVersion
+			} else {
+				// We track non-existent resources for non-wildcard streams until the client explicitly unsubscribes from them.
+				nextVersionMap[name] = ""
 			}
 		}
 	}
 
 	// Compute resources for removal regardless of the request type
-	for name := range state.GetResourceVersions() {
-		if _, ok := resources.resourceMap[name]; !ok {
+	for name, prevVersion := range state.GetResourceVersions() {
+		// The prevVersion != "" check is in place to make sure we are only sending an update to the client once right after it is removed.
+		// If the client decides to keep the subscription we skip the add for every subsequent response.
+		if _, ok := resources.resourceMap[name]; !ok && prevVersion != "" {
 			toRemove = append(toRemove, name)
 		}
 	}

pkg/cache/v3/linear_test.go

@@ -438,7 +438,7 @@ func TestLinearDeltaExistingResources(t *testing.T) {
 	w := make(chan DeltaResponse, 1)
 	c.CreateDeltaWatch(&DeltaRequest{TypeUrl: testType}, state, w)
 	checkDeltaWatchCount(t, c, 0)
-	verifyDeltaResponse(t, w, []resourceInfo{{"b", hashB}}, []string{"c"})
+	verifyDeltaResponse(t, w, []resourceInfo{{"b", hashB}}, []string{})
 
 	state = stream.NewStreamState(false, map[string]string{"a": "", "b": ""})
 	w = make(chan DeltaResponse, 1)