Add a test for blocking query on non-existent entry

This test shows how blocking queries are not efficient when the query
returns no results.  The test fails with 100+ calls instead of the
expected 2.

This test is still a bit flaky because it depends on the timing of the
writes. It can sometimes return 3 calls.

A future commit should fix this and make blocking queries even more
optimal for not-found results.

agent/consul/config_endpoint_test.go

@@ -1,6 +1,7 @@
 package consul
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"sort"
@@ -9,6 +10,7 @@ import (
 
 	msgpackrpc "github.com/hashicorp/consul-net-rpc/net-rpc-msgpackrpc"
 	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/errgroup"
 
 	"github.com/hashicorp/consul/acl"
 	"github.com/hashicorp/consul/agent/structs"
@@ -302,6 +304,67 @@ func TestConfigEntry_Get(t *testing.T) {
 	require.Equal(t, structs.ServiceDefaults, serviceConf.Kind)
 }
 
+func TestConfigEntry_Get_BlockOnNonExistent(t *testing.T) {
+	if testing.Short() {
+		t.Skip("too slow for testing.Short")
+	}
+
+	_, s1 := testServerWithConfig(t)
+	codec := rpcClient(t, s1)
+	store := s1.fsm.State()
+
+	entry := &structs.ServiceConfigEntry{
+		Kind: structs.ServiceDefaults,
+		Name: "alpha",
+	}
+	require.NoError(t, store.EnsureConfigEntry(1, entry))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	var count int
+
+	g, ctx := errgroup.WithContext(ctx)
+	g.Go(func() error {
+		args := structs.ConfigEntryQuery{
+			Kind: structs.ServiceDefaults,
+			Name: "does-not-exist",
+		}
+		args.QueryOptions.MaxQueryTime = time.Second
+
+		for ctx.Err() == nil {
+			var out structs.ConfigEntryResponse
+
+			err := msgpackrpc.CallWithCodec(codec, "ConfigEntry.Get", &args, &out)
+			if err != nil {
+				return err
+			}
+			t.Log("blocking query index", out.QueryMeta.Index, out.Entry)
+			count++
+			args.QueryOptions.MinQueryIndex = out.QueryMeta.Index
+		}
+		return nil
+	})
+
+	g.Go(func() error {
+		for i := uint64(0); i < 200; i++ {
+			time.Sleep(5 * time.Millisecond)
+			entry := &structs.ServiceConfigEntry{
+				Kind: structs.ServiceDefaults,
+				Name: fmt.Sprintf("other%d", i),
+			}
+			if err := store.EnsureConfigEntry(i+2, entry); err != nil {
+				return err
+			}
+		}
+		cancel()
+		return nil
+	})
+
+	require.NoError(t, g.Wait())
+	require.Equal(t, 2, count)
+}
+
 func TestConfigEntry_Get_ACLDeny(t *testing.T) {
 	if testing.Short() {
 		t.Skip("too slow for testing.Short")