Fix namespaced peer service updates / deletes.

This change fixes a function so that namespaced services are
correctly queried when handling updates / deletes. Prior to this
change, some peered services would not correctly be un-exported.

agent/grpc-external/services/peerstream/replication.go

@@ -285,9 +285,11 @@ func (s *Server) handleUpsertExportedServiceList(
 		exportedServices[snSidecarProxy] = struct{}{}
 		serviceNames = append(serviceNames, sn)
 	}
-	entMeta := structs.NodeEnterpriseMetaInPartition(partition)
 
-	_, serviceList, err := s.GetStore().ServiceList(nil, entMeta, peerName)
+	// Ensure we query services from all namespaces in this partition when we perform
+	// this query or else we may not propagate updates / deletes correctly.
+	entMeta := acl.NewEnterpriseMetaWithPartition(partition, acl.WildcardName)
+	_, serviceList, err := s.GetStore().ServiceList(nil, &entMeta, peerName)
 	if err != nil {
 		return err
 	}

agent/grpc-external/services/peerstream/stream_test.go

@@ -1960,7 +1960,7 @@ func processResponse_ExportedServiceUpdates(
 	localEntMeta acl.EnterpriseMeta,
 	peerName string,
 	tests []PeeringProcessResponse_testCase,
-) {
+) *MutableStatus {
 	// create a peering in the state store
 	peerID := "1fabcd52-1d46-49b0-b1d8-71559aee47f5"
 	require.NoError(t, store.PeeringWrite(31, &pbpeering.PeeringWriteRequest{
@@ -2041,6 +2041,7 @@ func processResponse_ExportedServiceUpdates(
 			run(t, tc)
 		})
 	}
+	return mst
 }
 
 func Test_processResponse_ExportedServiceUpdates(t *testing.T) {