Make blockingQuery efficient with 'not found' results.

By using the query results as state.

Blocking queries are efficient when the query matches some results,
because the ModifyIndex of those results, returned as queryMeta.Mindex,
will never change unless the items themselves change.

Blocking queries for non-existent items are not efficient because the
queryMeta.Index can (and often does) change when other entities are
written.

This commit reduces the churn of these queries by using a different
comparison for "has changed". Instead of using the modified index, we
use the existence of the results. If the previous result was "not found"
and the new result is still "not found", we know we can ignore the
modified index and continue to block.

This is done by setting the minQueryIndex to the returned
queryMeta.Index, which prevents the query from returning before a state
change is observed.

agent/consul/config_endpoint.go

@@ -142,7 +142,7 @@ func (c *ConfigEntry) Get(args *structs.ConfigEntryQuery, reply *structs.ConfigE
 
 			reply.Index = index
 			if entry == nil {
-				return nil
+				return errNotFound
 			}
 
 			reply.Entry = entry

agent/consul/rpc.go

@@ -935,6 +935,9 @@ func (s *Server) blockingQuery(queryOpts structs.QueryOptionsCompat, queryMeta s
 		var ws memdb.WatchSet
 		err := fn(ws, s.fsm.State())
 		s.setQueryMeta(queryMeta, queryOpts.GetToken())
+		if errors.Is(err, errNotFound) {
+			return nil
+		}
 		return err
 	}
 
@@ -947,6 +950,8 @@ func (s *Server) blockingQuery(queryOpts structs.QueryOptionsCompat, queryMeta s
 	// decrement the count when the function returns.
 	defer atomic.AddUint64(&s.queriesBlocking, ^uint64(0))
 
+	var notFound bool
+
 	for {
 		if queryOpts.GetRequireConsistent() {
 			if err := s.consistentRead(); err != nil {
@@ -966,7 +971,15 @@ func (s *Server) blockingQuery(queryOpts structs.QueryOptionsCompat, queryMeta s
 
 		err := fn(ws, state)
 		s.setQueryMeta(queryMeta, queryOpts.GetToken())
-		if err != nil {
+		switch {
+		case errors.Is(err, errNotFound):
+			if notFound {
+				// query result has not changed
+				minQueryIndex = queryMeta.GetIndex()
+			}
+
+			notFound = true
+		case err != nil:
 			return err
 		}
 
@@ -989,6 +1002,8 @@ func (s *Server) blockingQuery(queryOpts structs.QueryOptionsCompat, queryMeta s
 	}
 }
 
+var errNotFound = fmt.Errorf("no data found for query")
+
 // setQueryMeta is used to populate the QueryMeta data for an RPC call
 //
 // Note: This method must be called *after* filtering query results with ACLs.