chore: improve stability of flaky tests (#3596)

- add retry to mediator.Service.GetConnections
- add retry to ld/store.RemoteProviderStore.Save

Signed-off-by: Filip Burlacu <Filip.Burlacu@gendigital.com>

component/models/go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/VictoriaMetrics/fastcache v1.5.7
 	github.com/btcsuite/btcd v0.22.0-beta
 	github.com/btcsuite/btcutil v1.0.3-0.20201208143702-a53e38424cce
+	github.com/cenkalti/backoff/v4 v4.0.2
 	github.com/go-jose/go-jose/v3 v3.0.1-0.20221117193127-916db76e8214
 	github.com/google/tink/go v1.7.0
 	github.com/google/uuid v1.3.0

component/models/go.sum

@@ -25,6 +25,8 @@ github.com/btcsuite/snappy-go v0.0.0-20151229074030-0bdef8d06723/go.mod h1:8woku
 github.com/btcsuite/snappy-go v1.0.0/go.mod h1:8woku9dyThutzjeg+3xrA5iCpBRH8XEEg3lh6TiUghc=
 github.com/btcsuite/websocket v0.0.0-20150119174127-31079b680792/go.mod h1:ghJtEyQwv5/p4Mg4C0fgbePVuGr935/5ddU9Z3TmDRY=
 github.com/btcsuite/winsvc v1.0.0/go.mod h1:jsenWakMcC0zFBFurPLEAyrnc/teJEM1O46fmI40EZs=
+github.com/cenkalti/backoff/v4 v4.0.2 h1:JIufpQLbh4DkbQoii76ItQIUFzevQSqOLZca4eamEDs=
+github.com/cenkalti/backoff/v4 v4.0.2/go.mod h1:eEew/i+1Q6OrCDZh3WiXYv3+nJwBASZ8Bog/87DQnVg=
 github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+qY=
 github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/consensys/bavard v0.1.13 h1:oLhMLOFGTLdlda/kma4VOJazblc7IM5y5QPd2A/YjhQ=

component/models/ld/store/remote_provider_store.go

@@ -8,7 +8,9 @@ package store
 
 import (
 	"fmt"
+	"time"
 
+	"github.com/cenkalti/backoff/v4"
 	"github.com/google/uuid"
 
 	"github.com/hyperledger/aries-framework-go/spi/storage"
@@ -20,6 +22,9 @@ const (
 
 	// RemoteProviderRecordTag is a tag associated with every record in the store.
 	RemoteProviderRecordTag = "record"
+
+	queryRetryBackoffInterval = 100 * time.Millisecond
+	queryRetryMaxRetries      = 5
 )
 
 // RemoteProviderRecord is a record in store with remote provider info.
@@ -38,7 +43,8 @@ type RemoteProviderStore interface {
 
 // RemoteProviderStoreImpl is a default implementation of remote provider repository.
 type RemoteProviderStoreImpl struct {
-	store storage.Store
+	store               storage.Store
+	debugDisableBackoff bool
 }
 
 // NewRemoteProviderStore returns a new instance of RemoteProviderStoreImpl.
@@ -116,44 +122,14 @@ func (s *RemoteProviderStoreImpl) GetAll() ([]RemoteProviderRecord, error) {
 
 // Save creates a new remote provider record and saves it to the underlying storage.
 // If record with given endpoint already exists in the store, it is returned to the caller.
-func (s *RemoteProviderStoreImpl) Save(endpoint string) (*RemoteProviderRecord, error) { //nolint:gocyclo
-	iter, err := s.store.Query(RemoteProviderRecordTag)
+func (s *RemoteProviderStoreImpl) Save(endpoint string) (*RemoteProviderRecord, error) {
+	foundRecord, err := s.findEndpoint(endpoint)
 	if err != nil {
-		return nil, fmt.Errorf("query store: %w", err)
+		return nil, err
 	}
 
-	defer func() {
-		er := iter.Close()
-		if er != nil {
-			logger.Errorf("Failed to close iterator: %s", er.Error())
-		}
-	}()
-
-	for {
-		if ok, err := iter.Next(); !ok || err != nil {
-			if err != nil {
-				return nil, fmt.Errorf("next entry: %w", err)
-			}
-
-			break
-		}
-
-		k, err := iter.Key()
-		if err != nil {
-			return nil, fmt.Errorf("get key: %w", err)
-		}
-
-		v, err := iter.Value()
-		if err != nil {
-			return nil, fmt.Errorf("get value: %w", err)
-		}
-
-		if endpoint == string(v) {
-			return &RemoteProviderRecord{
-				ID:       k,
-				Endpoint: string(v),
-			}, nil
-		}
+	if foundRecord != nil {
+		return foundRecord, nil
 	}
 
 	record := &RemoteProviderRecord{
@@ -171,6 +147,64 @@ func (s *RemoteProviderStoreImpl) Save(endpoint string) (*RemoteProviderRecord,
 	return record, nil
 }
 
+func (s RemoteProviderStoreImpl) findEndpoint(endpoint string) (*RemoteProviderRecord, error) { // nolint:gocyclo
+	var (
+		foundRecord *RemoteProviderRecord
+		retries     backoff.BackOff
+	)
+
+	if s.debugDisableBackoff {
+		retries = &backoff.StopBackOff{}
+	} else {
+		retries = backoff.WithMaxRetries(backoff.NewConstantBackOff(queryRetryBackoffInterval), queryRetryMaxRetries)
+	}
+
+	return foundRecord, backoff.Retry(func() error {
+		iter, err := s.store.Query(RemoteProviderRecordTag)
+		if err != nil {
+			return fmt.Errorf("query store: %w", err)
+		}
+
+		defer func() {
+			er := iter.Close()
+			if er != nil {
+				logger.Errorf("Failed to close iterator: %s", er.Error())
+			}
+		}()
+
+		for {
+			if ok, err := iter.Next(); !ok || err != nil {
+				if err != nil {
+					return fmt.Errorf("next entry: %w", err)
+				}
+
+				break
+			}
+
+			k, err := iter.Key()
+			if err != nil {
+				return fmt.Errorf("get key: %w", err)
+			}
+
+			v, err := iter.Value()
+			if err != nil {
+				return fmt.Errorf("get value: %w", err)
+			}
+
+			if endpoint == string(v) {
+				foundRecord = &RemoteProviderRecord{
+					ID:       k,
+					Endpoint: string(v),
+				}
+
+				return nil
+			}
+		}
+
+		return nil
+	}, retries)
+}
+
 // Delete deletes a remote provider record in the underlying storage.
 func (s *RemoteProviderStoreImpl) Delete(id string) error {
 	if err := s.store.Delete(id); err != nil {