[FABG-812] Selection handling of 'access denied' error

When an 'access denied' error is received from the Discovery
server, close the selection service reference so that periodic
refresh is halted.

Change-Id: Id1156325147932e9fb085f0a76235cd404798b47
Signed-off-by: Bob Stasyszyn <Bob.Stasyszyn@securekey.com>

Author: bstasyszyn

pkg/client/common/discovery/dynamicdiscovery/service.go

@@ -69,9 +69,7 @@ func newService(config fab.EndpointConfig, query queryPeers, opts ...coptions.Op
 			if err != nil {
 				derr, ok := err.(*discoveryError)
 				if ok && derr.IsFatal() {
-					logger.Warnf("Got fatal error [%s]. Closing discovery client.", err)
-					s.lastErr = err
-					go func() { s.Close() }()
+					go s.close(err)
 				}
 			}
 			return peers, err
@@ -110,11 +108,26 @@ func (s *service) Close() {
 	s.peersRef.Close()
 }
 
+func (s *service) close(err error) {
+	logger.Warnf("Got fatal error [%s]. Closing discovery client.", err)
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	s.lastErr = err
+	s.peersRef.Close()
+}
+
+func (s *service) getLastError() error {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	return s.lastErr
+}
+
 // GetPeers returns the available peers
 func (s *service) GetPeers() ([]fab.Peer, error) {
 	if s.peersRef.IsClosed() {
-		if s.lastErr != nil {
-			return nil, errors.Errorf("Discovery client has been closed due to error: %s", s.lastErr)
+		lastErr := s.getLastError()
+		if lastErr != nil {
+			return nil, errors.Errorf("Discovery client has been closed due to error: %s", lastErr)
 		}
 		return nil, errors.Errorf("Discovery client has been closed")
 	}

pkg/client/common/selection/fabricselection/fabricselection.go

@@ -11,6 +11,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"strings"
+	"sync"
 	"time"
 
 	discclient "github.com/hyperledger/fabric-sdk-go/internal/github.com/hyperledger/fabric/discovery/client"
@@ -32,7 +33,10 @@ import (
 	grpcCodes "google.golang.org/grpc/codes"
 )
 
-const moduleName = "fabsdk/client"
+const (
+	moduleName   = "fabsdk/client"
+	accessDenied = "access denied"
+)
 
 var logger = logging.NewLogger(moduleName)
 
@@ -53,12 +57,13 @@ var defaultRetryOpts = retry.Opts{
 	RetryableCodes: retryableCodes,
 }
 
-type discoveryClient interface {
+// DiscoveryClient is the client to the discovery service
+type DiscoveryClient interface {
 	Send(ctx context.Context, req *discclient.Request, targets ...fab.PeerConfig) ([]fabdiscovery.Response, error)
 }
 
 // clientProvider is overridden by unit tests
-var clientProvider = func(ctx contextAPI.Client) (discoveryClient, error) {
+var clientProvider = func(ctx contextAPI.Client) (DiscoveryClient, error) {
 	return fabdiscovery.New(ctx)
 }
 
@@ -69,9 +74,11 @@ type Service struct {
 	responseTimeout time.Duration
 	ctx             contextAPI.Client
 	discovery       fab.DiscoveryService
-	discClient      discoveryClient
+	discClient      DiscoveryClient
 	chResponseCache *lazycache.Cache
 	retryOpts       retry.Opts
+	lastErr         error
+	lock            sync.RWMutex
 }
 
 // New creates a new dynamic selection service using Fabric's Discovery Service
@@ -123,7 +130,14 @@ func New(ctx contextAPI.Client, channelID string, discovery fab.DiscoveryService
 				logger.Debugf("Overriding retry opts: %#v", ropts)
 			}
 
-			return s.queryEndorsers(invocationChain, ropts)
+			endorsers, err := s.queryEndorsers(invocationChain, ropts)
+			if err != nil {
+				derr, ok := err.(discoveryError)
+				if ok && derr.isFatal() {
+					go s.close(err)
+				}
+			}
+			return endorsers, err
 		},
 		lazyref.WithRefreshInterval(lazyref.InitImmediately, options.refreshInterval),
 	)
@@ -133,6 +147,14 @@ func New(ctx contextAPI.Client, channelID string, discovery fab.DiscoveryService
 
 // GetEndorsersForChaincode returns the endorsing peers for the given chaincodes
 func (s *Service) GetEndorsersForChaincode(chaincodes []*fab.ChaincodeCall, opts ...coptions.Opt) ([]fab.Peer, error) {
+	if s.chResponseCache.IsClosed() {
+		lastErr := s.getLastError()
+		if lastErr != nil {
+			return nil, errors.Errorf("Selection service has been closed due to error: %s", lastErr)
+		}
+		return nil, errors.Errorf("Selection service has been closed")
+	}
+
 	logger.Debugf("Getting endorsers for chaincodes [%#v]...", chaincodes)
 	if len(chaincodes) == 0 {
 		return nil, errors.New("no chaincode IDs provided")
@@ -167,6 +189,20 @@ func (s *Service) Close() {
 	s.chResponseCache.Close()
 }
 
+func (s *Service) close(err error) {
+	logger.Warnf("Got fatal error [%s]. Closing selection service.", err)
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	s.lastErr = err
+	s.chResponseCache.Close()
+}
+
+func (s *Service) getLastError() error {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	return s.lastErr
+}
+
 func (s *Service) getEndorsers(chaincodes []*fab.ChaincodeCall, chResponse discclient.ChannelResponse, peerFilter soptions.PeerFilter, sorter soptions.PeerSorter) (discclient.Endorsers, error) {
 	peers, err := s.discovery.GetPeers()
 	if err != nil {
@@ -214,9 +250,9 @@ func (s *Service) queryEndorsers(chaincodes []*fab.ChaincodeCall, retryOpts retr
 	)
 
 	if err != nil {
-		return nil, err
+		return nil, newDiscoveryError(err)
 	}
-	return chResponse.(discclient.ChannelResponse), err
+	return chResponse.(discclient.ChannelResponse), nil
 }
 
 func (s *Service) query(req *discclient.Request, chaincodes []*fab.ChaincodeCall, targets []fab.PeerConfig) (discclient.ChannelResponse, error) {
@@ -355,3 +391,7 @@ func (e discoveryError) isTransient() bool {
 	return strings.Contains(e.Error(), "failed constructing descriptor for chaincodes") ||
 		strings.Contains(e.Error(), "no endorsement combination can be satisfied")
 }
+
+func (e discoveryError) isFatal() bool {
+	return e == accessDenied
+}

pkg/client/common/selection/fabricselection/selection_test.go

@@ -1,3 +1,5 @@
+// +build testing
+
 /*
 Copyright SecureKey Technologies Inc. All Rights Reserved.
 
@@ -114,15 +116,15 @@ func TestSelection(t *testing.T) {
 
 	discClient := clientmocks.NewMockDiscoveryClient()
 
-	clientProvider = func(ctx contextAPI.Client) (discoveryClient, error) {
+	SetClientProvider(func(ctx contextAPI.Client) (DiscoveryClient, error) {
 		return discClient, nil
-	}
+	})
 
 	service, err := New(
 		ctx, channelID,
 		mocks.NewMockDiscoveryService(nil, peer1Org1, peer2Org1, peer1Org2, peer2Org2, peer1Org3, peer2Org3),
-		WithRefreshInterval(100*time.Millisecond),
-		WithResponseTimeout(10*time.Millisecond),
+		WithRefreshInterval(10*time.Millisecond),
+		WithResponseTimeout(100*time.Millisecond),
 	)
 	require.NoError(t, err)
 	defer service.Close()
@@ -135,7 +137,7 @@ func TestSelection(t *testing.T) {
 				Error:         fmt.Errorf("simulated response error"),
 			},
 		)
-		testSelectionError(t, service)
+		testSelectionError(t, service, "error getting channel response for channel [testchannel]: simulated response error")
 	})
 
 	t.Run("CCtoCC", func(t *testing.T) {
@@ -149,7 +151,7 @@ func TestSelection(t *testing.T) {
 		)
 
 		// Wait for cache to refresh
-		time.Sleep(200 * time.Millisecond)
+		time.Sleep(20 * time.Millisecond)
 		testSelectionCCtoCC(t, service)
 	})
 
@@ -168,6 +170,18 @@ func TestSelection(t *testing.T) {
 	t.Run("Priority Selector", func(t *testing.T) {
 		testSelectionPrioritySelector(t, service)
 	})
+
+	t.Run("Fatal Error", func(t *testing.T) {
+		discClient.SetResponses(
+			&clientmocks.MockDiscoverEndpointResponse{
+				PeerEndpoints: []*discmocks.MockDiscoveryPeerEndpoint{},
+				Error:         fmt.Errorf(accessDenied),
+			},
+		)
+		// Wait for cache to refresh
+		time.Sleep(20 * time.Millisecond)
+		testSelectionError(t, service, "Selection service has been closed due to error: access denied")
+	})
 }
 
 func TestWithDiscoveryFilter(t *testing.T) {
@@ -179,9 +193,9 @@ func TestWithDiscoveryFilter(t *testing.T) {
 	ctx.SetEndpointConfig(config)
 
 	discClient := clientmocks.NewMockDiscoveryClient()
-	clientProvider = func(ctx contextAPI.Client) (discoveryClient, error) {
+	SetClientProvider(func(ctx contextAPI.Client) (DiscoveryClient, error) {
 		return discClient, nil
-	}
+	})
 
 	discClient.SetResponses(
 		&clientmocks.MockDiscoverEndpointResponse{
@@ -241,9 +255,10 @@ func TestWithDiscoveryFilter(t *testing.T) {
 	})
 }
 
-func testSelectionError(t *testing.T, service *Service) {
+func testSelectionError(t *testing.T, service *Service, expectedErrMsg string) {
 	endorsers, err := service.GetEndorsersForChaincode([]*fab.ChaincodeCall{{ID: cc1}})
 	assert.Error(t, err)
+	assert.Equal(t, expectedErrMsg, err.Error())
 	assert.Equal(t, 0, len(endorsers))
 }
 

pkg/client/common/selection/fabricselection/selection_test_env.go

@@ -0,0 +1,18 @@
+// +build testing
+
+/*
+Copyright SecureKey Technologies Inc. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package fabricselection
+
+import (
+	contextAPI "github.com/hyperledger/fabric-sdk-go/pkg/common/providers/context"
+)
+
+// SetClientProvider overrides the discovery client provider for unit tests
+func SetClientProvider(provider func(ctx contextAPI.Client) (DiscoveryClient, error)) {
+	clientProvider = provider
+}

pkg/fabsdk/provider/chpvdr/chprovider_test.go

@@ -11,6 +11,7 @@ package chpvdr
 import (
 	"errors"
 	"testing"
+	"time"
 
 	"github.com/hyperledger/fabric-sdk-go/pkg/client/common/discovery/dynamicdiscovery"
 	"github.com/hyperledger/fabric-sdk-go/pkg/client/common/discovery/staticdiscovery"
@@ -115,7 +116,69 @@ func TestBasicValidChannel(t *testing.T) {
 	assert.Truef(t, ok, "Expecting selection to be Fabric for v1_2")
 }
 
-func TestAccessDenied(t *testing.T) {
+func TestDiscoveryAccessDenied(t *testing.T) {
+	discClient, channelService := setupDiscovery(t, func(discClient *clientmocks.MockDiscoveryClient) {
+		dynamicdiscovery.SetClientProvider(func(ctx context.Client) (dynamicdiscovery.DiscoveryClient, error) {
+			return discClient, nil
+		})
+	})
+
+	discClient.SetResponses(
+		&clientmocks.MockDiscoverEndpointResponse{
+			Error: errors.New("access denied"),
+		},
+	)
+
+	discovery, err := channelService.Discovery()
+	require.NoError(t, err)
+	require.NotNil(t, discovery)
+	_, ok := discovery.(*dynamicdiscovery.ChannelService)
+	assert.Truef(t, ok, "Expecting discovery to be Dynamic for v1_2")
+
+	_, err = discovery.GetPeers()
+	require.Error(t, err)
+	assert.Equal(t, "access denied", err.Error())
+
+	time.Sleep(50 * time.Millisecond)
+
+	// Subsequent calls should fail since the service is closed
+	_, err = discovery.GetPeers()
+	require.Error(t, err)
+	assert.Equal(t, "Discovery client has been closed due to error: access denied", err.Error())
+}
+
+func TestSelectionAccessDenied(t *testing.T) {
+	discClient, channelService := setupDiscovery(t, func(discClient *clientmocks.MockDiscoveryClient) {
+		fabricselection.SetClientProvider(func(ctx context.Client) (fabricselection.DiscoveryClient, error) {
+			return discClient, nil
+		})
+	})
+
+	discClient.SetResponses(
+		&clientmocks.MockDiscoverEndpointResponse{
+			Error: errors.New("access denied"),
+		},
+	)
+
+	selection, err := channelService.Selection()
+	require.NoError(t, err)
+	require.NotNil(t, selection)
+	_, ok := selection.(*fabricselection.Service)
+	assert.Truef(t, ok, "Expecting selection to be Fabric for v1_2")
+
+	_, err = selection.GetEndorsersForChaincode([]*fab.ChaincodeCall{{ID: "cc1"}})
+	require.Error(t, err)
+	assert.Equal(t, "error getting channel response for channel [testchannel]: access denied", err.Error())
+
+	time.Sleep(50 * time.Millisecond)
+
+	// Subsequent calls should fail since the service is closed
+	_, err = selection.GetEndorsersForChaincode([]*fab.ChaincodeCall{{ID: "cc1"}})
+	require.Error(t, err)
+	assert.Equal(t, "Selection service has been closed due to error: access denied", err.Error())
+}
+
+func setupDiscovery(t *testing.T, preInit func(discClient *clientmocks.MockDiscoveryClient)) (*clientmocks.MockDiscoveryClient, fab.ChannelService) {
 	ctx := mocks.NewMockProviderContext()
 
 	user := mspmocks.NewMockSigningIdentity("user", "user")
@@ -127,15 +190,7 @@ func TestAccessDenied(t *testing.T) {
 
 	discClient := clientmocks.NewMockDiscoveryClient()
 
-	discClient.SetResponses(
-		&clientmocks.MockDiscoverEndpointResponse{
-			Error: errors.New("access denied"),
-		},
-	)
-
-	dynamicdiscovery.SetClientProvider(func(ctx context.Client) (dynamicdiscovery.DiscoveryClient, error) {
-		return discClient, nil
-	})
+	preInit(discClient)
 
 	cp, err := New(clientCtx.EndpointConfig())
 	require.NoError(t, err)
@@ -152,13 +207,5 @@ func TestAccessDenied(t *testing.T) {
 	channelService, err := cp.ChannelService(clientCtx, "testchannel")
 	require.NoError(t, err)
 
-	discovery, err := channelService.Discovery()
-	require.NoError(t, err)
-	require.NotNil(t, discovery)
-	_, ok := discovery.(*dynamicdiscovery.ChannelService)
-	assert.Truef(t, ok, "Expecting discovery to be Dynamic for v1_2")
-
-	_, err = discovery.GetPeers()
-	require.Error(t, err)
-	assert.Equal(t, "access denied", err.Error())
+	return discClient, channelService
 }

pkg/util/concurrent/lazycache/lazycache.go

@@ -169,6 +169,11 @@ func (c *Cache) Close() {
 	c.DeleteAll()
 }
 
+// IsClosed reeturns true if the cache has been closed
+func (c *Cache) IsClosed() bool {
+	return atomic.LoadInt32(&c.closed) == 1
+}
+
 // DeleteAll does the following:
 // - calls Close on all values that implement a Close() function
 // - deletes all entries from the cache

pkg/util/concurrent/lazycache/lazycache_test.go

@@ -261,6 +261,8 @@ func TestClose(t *testing.T) {
 	// Close the cache
 	cache.Close()
 
+	assert.True(t, cache.IsClosed())
+
 	// Close again should be fine
 	cache.Close()