health: ensure /v1/health/service/:service endpoint returns the most recent results when a filter is used

This is two bugs in two subsystems (blocking queries ; streaming) that
make the overall endpoint behave brokenly in the same way to the end
user.

Simple reproduction (streaming):

1. Register a service with a tag.

    curl -sL --request PUT 'http://localhost:8500/v1/agent/service/register' \
        --header 'Content-Type: application/json' \
        --data-raw '{ "ID": "ID1", "Name": "test", "Tags":[ "a" ], "EnableTagOverride": true }'

2. Do an initial filter query that matches on the tag.

    curl -sLi --get 'http://localhost:8500/v1/health/service/test' --data-urlencode 'filter=a in Service.Tags'

3. Note you get one result. Use the `X-Consul-Index` header to establish
   a blocking query in another terminal, this should not return yet.

    curl -sLi --get 'http://localhost:8500/v1/health/service/test?index=$INDEX' --data-urlencode 'filter=a in Service.Tags'

4. Re-register that service with a different tag.

    curl -sL --request PUT 'http://localhost:8500/v1/agent/service/register' \
        --header 'Content-Type: application/json' \
        --data-raw '{ "ID": "ID1", "Name": "test", "Tags":[ "b" ], "EnableTagOverride": true }'

5. Your blocking query from (3) should return with a header
   `X-Consul-Query-Backend: streaming` and empty results if it works
   correctly `[]`.

To reproduce for non-streaming, simply add `&near=_agent` to your read
queries and ensure `X-Consul-Query-Backend: blocking-query` shows up in the results.

Author: rboyer

Diff for: agent/consul/health_endpoint.go

@@ -236,21 +236,22 @@ func (h *Health) ServiceNodes(args *structs.ServiceSpecificRequest, reply *struc
 		&args.QueryOptions,
 		&reply.QueryMeta,
 		func(ws memdb.WatchSet, state *state.Store) error {
+			var thisReply structs.IndexedCheckServiceNodes
+
 			index, nodes, err := f(ws, state, args)
 			if err != nil {
 				return err
 			}
 
-			reply.Index, reply.Nodes = index, nodes
 			if len(args.NodeMetaFilters) > 0 {
-				reply.Nodes = nodeMetaFilter(args.NodeMetaFilters, reply.Nodes)
+				nodes = nodeMetaFilter(args.NodeMetaFilters, nodes)
 			}
 
-			raw, err := filter.Execute(reply.Nodes)
+			raw, err := filter.Execute(nodes)
 			if err != nil {
 				return err
 			}
-			reply.Nodes = raw.(structs.CheckServiceNodes)
+			nodes = raw.(structs.CheckServiceNodes)
 
 			// Note: we filter the results with ACLs *after* applying the user-supplied
 			// bexpr filter, to ensure QueryMeta.ResultsFilteredByACLs does not include
@@ -259,7 +260,12 @@ func (h *Health) ServiceNodes(args *structs.ServiceSpecificRequest, reply *struc
 				return err
 			}
 
-			return h.srv.sortNodesByDistanceFrom(args.Source, reply.Nodes)
+			if err := h.srv.sortNodesByDistanceFrom(args.Source, nodes); err != nil {
+				return err
+			}
+
+			reply.Index, reply.Nodes = index, nodes
+			return nil
 		})
 
 	// Provide some metrics

Diff for: agent/rpcclient/health/view.go

@@ -80,11 +80,12 @@ func (s *healthView) Update(events []*pbsubscribe.Event) error {
 				return errors.New("check service node was unexpectedly nil")
 			}
 			passed, err := s.filter.Evaluate(*csn)
-			switch {
-			case err != nil:
+			if err != nil {
 				return err
-			case passed:
+			} else if passed {
 				s.state[id] = *csn
+			} else {
+				delete(s.state, id)
 			}
 
 		case pbsubscribe.CatalogOp_Deregister: