state: ensure we unblock intentions queries upon the upgrade to config entries (#9062)

1. do a state store query to list intentions as the agent would do over in `agent/proxycfg` backing `agent/xds`
2. upgrade the database and do a fresh `service-intentions` config entry write
3. the blocking query inside of the agent cache in (1) doesn't notice (2)

agent/consul/state/intention.go

@@ -140,11 +140,11 @@ func (s *Restore) LegacyIntention(ixn *structs.Intention) error {
 func (s *Store) AreIntentionsInConfigEntries() (bool, error) {
 	tx := s.db.Txn(false)
 	defer tx.Abort()
-	return areIntentionsInConfigEntries(tx)
+	return areIntentionsInConfigEntries(tx, nil)
 }
 
-func areIntentionsInConfigEntries(tx ReadTxn) (bool, error) {
-	_, entry, err := systemMetadataGetTxn(tx, nil, structs.SystemMetadataIntentionFormatKey)
+func areIntentionsInConfigEntries(tx ReadTxn, ws memdb.WatchSet) (bool, error) {
+	_, entry, err := systemMetadataGetTxn(tx, ws, structs.SystemMetadataIntentionFormatKey)
 	if err != nil {
 		return false, fmt.Errorf("failed system metadatalookup: %s", err)
 	}
@@ -169,7 +169,7 @@ func (s *Store) Intentions(ws memdb.WatchSet, entMeta *structs.EnterpriseMeta) (
 	tx := s.db.Txn(false)
 	defer tx.Abort()
 
-	usingConfigEntries, err := areIntentionsInConfigEntries(tx)
+	usingConfigEntries, err := areIntentionsInConfigEntries(tx, ws)
 	if err != nil {
 		return 0, nil, false, err
 	}
@@ -214,7 +214,7 @@ func (s *Store) LegacyIntentionSet(idx uint64, ixn *structs.Intention) error {
 	tx := s.db.WriteTxn(idx)
 	defer tx.Abort()
 
-	usingConfigEntries, err := areIntentionsInConfigEntries(tx)
+	usingConfigEntries, err := areIntentionsInConfigEntries(tx, nil)
 	if err != nil {
 		return err
 	}
@@ -291,7 +291,7 @@ func (s *Store) IntentionGet(ws memdb.WatchSet, id string) (uint64, *structs.Ser
 	tx := s.db.Txn(false)
 	defer tx.Abort()
 
-	usingConfigEntries, err := areIntentionsInConfigEntries(tx)
+	usingConfigEntries, err := areIntentionsInConfigEntries(tx, ws)
 	if err != nil {
 		return 0, nil, nil, err
 	}
@@ -330,7 +330,7 @@ func (s *Store) IntentionGetExact(ws memdb.WatchSet, args *structs.IntentionQuer
 	tx := s.db.Txn(false)
 	defer tx.Abort()
 
-	usingConfigEntries, err := areIntentionsInConfigEntries(tx)
+	usingConfigEntries, err := areIntentionsInConfigEntries(tx, ws)
 	if err != nil {
 		return 0, nil, nil, err
 	}
@@ -376,7 +376,7 @@ func (s *Store) LegacyIntentionDelete(idx uint64, id string) error {
 	tx := s.db.WriteTxn(idx)
 	defer tx.Abort()
 
-	usingConfigEntries, err := areIntentionsInConfigEntries(tx)
+	usingConfigEntries, err := areIntentionsInConfigEntries(tx, nil)
 	if err != nil {
 		return err
 	}
@@ -522,7 +522,7 @@ func (s *Store) IntentionMatch(ws memdb.WatchSet, args *structs.IntentionQueryMa
 	tx := s.db.Txn(false)
 	defer tx.Abort()
 
-	usingConfigEntries, err := areIntentionsInConfigEntries(tx)
+	usingConfigEntries, err := areIntentionsInConfigEntries(tx, ws)
 	if err != nil {
 		return 0, nil, err
 	}
@@ -571,7 +571,7 @@ func (s *Store) IntentionMatchOne(
 	tx := s.db.Txn(false)
 	defer tx.Abort()
 
-	usingConfigEntries, err := areIntentionsInConfigEntries(tx)
+	usingConfigEntries, err := areIntentionsInConfigEntries(tx, ws)
 	if err != nil {
 		return 0, nil, err
 	}

agent/consul/state/intention_test.go

@@ -986,6 +986,37 @@ func TestStore_IntentionMatchOne_table(t *testing.T) {
 	}
 }
 
+func TestStore_IntentionMatch_WatchesDuringUpgrade(t *testing.T) {
+	s := testStateStore(t)
+
+	args := structs.IntentionQueryMatch{
+		Type: structs.IntentionMatchDestination,
+		Entries: []structs.IntentionMatchEntry{
+			{Namespace: "default", Name: "api"},
+		},
+	}
+
+	// Start with an empty, un-upgraded database and do a watch.
+
+	ws := memdb.NewWatchSet()
+	_, matches, err := s.IntentionMatch(ws, &args)
+	require.NoError(t, err)
+	require.Len(t, matches, 1)    // one request gets one response
+	require.Len(t, matches[0], 0) // but no intentions
+
+	disableLegacyIntentions(s)
+	conf := &structs.ServiceIntentionsConfigEntry{
+		Kind: structs.ServiceIntentions,
+		Name: "api",
+		Sources: []*structs.SourceIntention{
+			{Name: "web", Action: structs.IntentionActionAllow},
+		},
+	}
+	require.NoError(t, s.EnsureConfigEntry(1, conf, &conf.EnterpriseMeta))
+
+	require.True(t, watchFired(ws))
+}
+
 func TestStore_LegacyIntention_Snapshot_Restore(t *testing.T) {
 	// note: irrelevant test for config entries variant
 	s := testStateStore(t)