Improve cluster metadata and metadata persistence tests (temporalio#7249)

## What changed?
Improved persistence tests to account for differing behavior of some
persistence backends.

## Why?
These tests made assumptions about persistence that don't necessarily
apply to all persistence backends. Specifically the tests assumed:
- All persistence backends return a final empty page when paginating
results
- All persistence backends are guaranteed to expire TTL'd records
(nearly) instantly upon expiration.

## How did you test it?
Ran the tests with various persistence backends

## Potential risks
This is a test-only update

## Documentation
No changes.

## Is hotfix candidate?
No

Author: jprieto-temporal

common/persistence/persistence-tests/cluster_metadata_manager.go

@@ -65,6 +65,9 @@ func (s *ClusterMetadataManagerSuite) SetupTest() {
 
 // TearDownTest implementation
 func (s *ClusterMetadataManagerSuite) TearDownTest() {
+	// Ensure all tests clean up after themselves
+	// Todo: MetaMgr should provide api to clear all members
+	s.waitForPrune(1 * time.Second)
 	s.cancel()
 }
 
@@ -100,16 +103,12 @@ func (s *ClusterMetadataManagerSuite) TestClusterMembershipUpsertCanReadAny() {
 	s.Nil(err)
 	s.NotNil(resp)
 	s.NotEmpty(resp.ActiveMembers)
+
+	s.waitForPrune(5 * time.Second)
 }
 
 // TestClusterMembershipUpsertCanRead verifies that we can UpsertClusterMembership and read our result
 func (s *ClusterMetadataManagerSuite) TestClusterMembershipUpsertCanPageRead() {
-	// Expire previous records
-	// Todo: MetaMgr should provide api to clear all members
-	time.Sleep(time.Second * 3)
-	err := s.ClusterMetadataManager.PruneClusterMembership(s.ctx, &p.PruneClusterMembershipRequest{MaxRecordsPruned: 100})
-	s.Nil(err)
-
 	expectedIds := make(map[string]int, 100)
 	for i := 0; i < 100; i++ {
 		hostID := primitives.NewUUID().Downcast()
@@ -148,9 +147,7 @@ func (s *ClusterMetadataManagerSuite) TestClusterMembershipUpsertCanPageRead() {
 		s.Zero(val, "identifier was either not found in db, or shouldn't be there - "+id)
 	}
 
-	time.Sleep(time.Second * 3)
-	err = s.ClusterMetadataManager.PruneClusterMembership(s.ctx, &p.PruneClusterMembershipRequest{MaxRecordsPruned: 1000})
-	s.NoError(err)
+	s.waitForPrune(5 * time.Second)
 }
 
 func (s *ClusterMetadataManagerSuite) validateUpsert(req *p.UpsertClusterMembershipRequest, resp *p.GetClusterMembersResponse, err error) {
@@ -223,10 +220,7 @@ func (s *ClusterMetadataManagerSuite) TestClusterMembershipReadFiltersCorrectly(
 	)
 
 	s.validateUpsert(req, resp, err)
-
-	time.Sleep(time.Second * 3)
-	err = s.ClusterMetadataManager.PruneClusterMembership(s.ctx, &p.PruneClusterMembershipRequest{MaxRecordsPruned: 1000})
-	s.NoError(err)
+	s.waitForPrune(5 * time.Second)
 }
 
 // TestClusterMembershipUpsertExpiresCorrectly verifies RecordExpiry functions properly for ClusterMembership records
@@ -263,19 +257,27 @@ func (s *ClusterMetadataManagerSuite) TestClusterMembershipUpsertExpiresCorrectl
 	s.Equal(resp.ActiveMembers[0].HostID, req.HostID)
 	s.Equal(resp.ActiveMembers[0].Role, req.Role)
 
-	time.Sleep(time.Second * 2)
-
-	err = s.ClusterMetadataManager.PruneClusterMembership(s.ctx, &p.PruneClusterMembershipRequest{MaxRecordsPruned: 100})
-	s.Nil(err)
+	s.waitForPrune(5 * time.Second)
+}
 
-	resp, err = s.ClusterMetadataManager.GetClusterMembers(
-		s.ctx,
-		&p.GetClusterMembersRequest{LastHeartbeatWithin: time.Minute * 10},
-	)
+// waitForPrune waits up for the persistence backend to prune all records. Some persistence backends
+// may not remove TTL'd entries at the exact instant they should expire, so we allow some timing flexibility here.
+func (s *ClusterMetadataManagerSuite) waitForPrune(waitFor time.Duration) {
+	s.Eventually(func() bool {
+		err := s.ClusterMetadataManager.PruneClusterMembership(s.ctx, &p.PruneClusterMembershipRequest{MaxRecordsPruned: 100})
+		s.Nil(err)
+
+		resp, err := s.ClusterMetadataManager.GetClusterMembers(
+			s.ctx,
+			&p.GetClusterMembersRequest{LastHeartbeatWithin: time.Minute * 10},
+		)
+		s.NoError(err)
+		s.NotNil(resp)
+		return len(resp.ActiveMembers) == 0
 
-	s.Nil(err)
-	s.NotNil(resp)
-	s.Empty(resp.ActiveMembers)
+	},
+		waitFor,
+		500*time.Millisecond)
 }
 
 // TestClusterMembershipUpsertInvalidExpiry verifies we cannot specify a non-positive RecordExpiry duration

common/persistence/persistence-tests/metadata_persistence_v2.go

@@ -1334,14 +1334,17 @@ func (m *MetadataPersistenceSuiteV2) TestListNamespaces() {
 			// so we can test == easily
 			namespace.NotificationVersion = 0
 		}
-		pageCount++
+		// Some persistence backends return an unavoidable empty final page.
+		if len(resp.Namespaces) > 0 {
+			pageCount++
+		}
 		if len(token) == 0 {
 			break
 		}
 	}
 
-	// 2 pages with data and 1 empty page which is unavoidable.
-	m.Equal(pageCount, 3)
+	// There should be 2 non-empty pages.
+	m.Equal(pageCount, 2)
 	m.Equal(len(inputNamespaces), len(outputNamespaces))
 	for _, namespace := range inputNamespaces {
 		m.DeepEqual(namespace, outputNamespaces[namespace.Namespace.Info.Id])
@@ -1415,14 +1418,17 @@ func (m *MetadataPersistenceSuiteV2) TestListNamespaces_DeletedNamespace() {
 		m.NoError(err)
 		token = resp.NextPageToken
 		listNamespacesPageSize2 = append(listNamespacesPageSize2, resp.Namespaces...)
-		pageCount++
+		// Some persistence backends return an unavoidable empty final page.
+		if len(resp.Namespaces) > 0 {
+			pageCount++
+		}
 		if len(token) == 0 {
 			break
 		}
 	}
 
-	// 1 page with data and 1 empty page which is unavoidable.
-	m.Equal(2, pageCount)
+	// There should be 1 non-empty page.
+	m.Equal(1, pageCount)
 	m.Len(listNamespacesPageSize2, 2)
 	for _, namespace := range listNamespacesPageSize2 {
 		m.NotEqual(namespace.Namespace.Info.State, enumspb.NAMESPACE_STATE_DELETED)
@@ -1435,14 +1441,17 @@ func (m *MetadataPersistenceSuiteV2) TestListNamespaces_DeletedNamespace() {
 		m.NoError(err)
 		token = resp.NextPageToken
 		listNamespacesPageSize1 = append(listNamespacesPageSize1, resp.Namespaces...)
-		pageCount++
+		// Some persistence backends return an unavoidable empty final page.
+		if len(resp.Namespaces) > 0 {
+			pageCount++
+		}
 		if len(token) == 0 {
 			break
 		}
 	}
 
-	// 2 pages with data and 1 empty page which is unavoidable.
-	m.Equal(3, pageCount)
+	// There should be 2 non-empty pages.
+	m.Equal(2, pageCount)
 	m.Len(listNamespacesPageSize1, 2)
 	for _, namespace := range listNamespacesPageSize1 {
 		m.NotEqual(namespace.Namespace.Info.State, enumspb.NAMESPACE_STATE_DELETED)