Merge develop into master (#390)

* fix builds

* rename COC for github (#382)

* Hardcode External IP (#386)

* Hardcode External IP

* Hardcode External IP

* Add Connect timeout to p2p conns (#387)

* Add Connect timeout to p2p conns

* add exported comment

* Added timeout for eth read operations (#388)

* Added timeout for read operations

* fix unit test

* manual group import

anchors/anchor_test.go

@@ -3,12 +3,25 @@
 package anchors
 
 import (
+	"os"
 	"testing"
 
+	"github.com/centrifuge/go-centrifuge/bootstrap"
+	"github.com/centrifuge/go-centrifuge/config"
 	"github.com/centrifuge/go-centrifuge/utils"
 	"github.com/stretchr/testify/assert"
 )
 
+func TestMain(m *testing.M) {
+	ibootstappers := []bootstrap.TestBootstrapper{
+		&config.Bootstrapper{},
+	}
+	bootstrap.RunTestBootstrappers(ibootstappers, nil)
+	result := m.Run()
+	bootstrap.RunTestTeardown(ibootstappers)
+	os.Exit(result)
+}
+
 func TestNewAnchorId(t *testing.T) {
 	tests := []struct {
 		name  string

anchors/ethereum_anchor_repository.go

@@ -48,7 +48,9 @@ func NewEthereumAnchorRepository(config Config, anchorRepositoryContract AnchorR
 
 // Commits takes an anchorID and returns the corresponding documentRoot from the chain
 func (ethRepository *EthereumAnchorRepository) GetDocumentRootOf(anchorID AnchorID) (docRoot DocRoot, err error) {
-	return ethRepository.anchorRepositoryContract.Commits(ethereum.GetGethCallOpts(), anchorID.BigInt())
+	// Ignoring cancelFunc as code will block until response or timeout is triggered
+	opts, _ := ethereum.GetGethCallOpts()
+	return ethRepository.anchorRepositoryContract.Commits(opts, anchorID.BigInt())
 }
 
 //PreCommitAnchor will call the transaction PreCommit on the smart contract

anchors/ethereum_anchor_repository_test.go

@@ -1,12 +1,11 @@
 // +build unit
 
-package anchors_test
+package anchors
 
 import (
 	"math/big"
 	"testing"
 
-	"github.com/centrifuge/go-centrifuge/anchors"
 	"github.com/centrifuge/go-centrifuge/config"
 	"github.com/centrifuge/go-centrifuge/identity"
 	"github.com/centrifuge/go-centrifuge/keytools/secp256k1"
@@ -19,7 +18,7 @@ import (
 
 type mockAnchorRepo struct {
 	mock.Mock
-	anchors.AnchorRepositoryContract
+	AnchorRepositoryContract
 }
 
 func (m *mockAnchorRepo) Commits(opts *bind.CallOpts, anchorID *big.Int) (docRoot [32]byte, err error) {
@@ -36,10 +35,10 @@ func TestCorrectCommitSignatureGen(t *testing.T) {
 	correctCommitToSign := "0x15f9cb57608a7ef31428fd6b1cb7ea2002ab032211d882b920c1474334004d6b"
 	correctCommitSignature := "0xb4051d6d03c3bf39f4ec4ba949a91a358b0cacb4804b82ed2ba978d338f5e747770c00b63c8e50c1a7aa5ba629870b54c2068a56f8b43460aa47891c6635d36d01"
 	testPrivateKey, _ := hexutil.Decode("0x17e063fa17dd8274b09c14b253697d9a20afff74ace3c04fdb1b9c814ce0ada5")
-	anchorIDTyped, _ := anchors.NewAnchorID(anchorID)
+	anchorIDTyped, _ := NewAnchorID(anchorID)
 	centIdTyped, _ := identity.ToCentID(centrifugeId)
-	docRootTyped, _ := anchors.NewDocRoot(documentRoot)
-	messageToSign := anchors.GenerateCommitHash(anchorIDTyped, centIdTyped, docRootTyped)
+	docRootTyped, _ := NewDocRoot(documentRoot)
+	messageToSign := GenerateCommitHash(anchorIDTyped, centIdTyped, docRootTyped)
 	assert.Equal(t, correctCommitToSign, hexutil.Encode(messageToSign), "messageToSign not calculated correctly")
 	signature, _ := secp256k1.SignEthereum(messageToSign, testPrivateKey)
 	assert.Equal(t, correctCommitSignature, hexutil.Encode(signature), "signature not correct")
@@ -55,16 +54,16 @@ func TestGenerateAnchor(t *testing.T) {
 	var documentProofs [][32]byte
 	documentProofs = append(documentProofs, documentProof)
 	centIdTyped, _ := identity.ToCentID(centrifugeId)
-	messageToSign := anchors.GenerateCommitHash(currentAnchorID, centIdTyped, currentDocumentRoot)
+	messageToSign := GenerateCommitHash(currentAnchorID, centIdTyped, currentDocumentRoot)
 	signature, _ := secp256k1.SignEthereum(messageToSign, testPrivateKey)
 
 	var documentRoot32Bytes [32]byte
 	copy(documentRoot32Bytes[:], currentDocumentRoot[:32])
 
-	commitData := anchors.NewCommitData(0, currentAnchorID, documentRoot32Bytes, centIdTyped, documentProofs, signature)
+	commitData := NewCommitData(0, currentAnchorID, documentRoot32Bytes, centIdTyped, documentProofs, signature)
 
-	anchorID, _ := anchors.NewAnchorID(currentAnchorID[:])
-	docRoot, _ := anchors.NewDocRoot(documentRoot32Bytes[:])
+	anchorID, _ := NewAnchorID(currentAnchorID[:])
+	docRoot, _ := NewDocRoot(documentRoot32Bytes[:])
 
 	assert.Equal(t, commitData.AnchorID, anchorID, "Anchor should have the passed ID")
 	assert.Equal(t, commitData.DocumentRoot, docRoot, "Anchor should have the passed document root")
@@ -75,10 +74,10 @@ func TestGenerateAnchor(t *testing.T) {
 
 func TestGetDocumentRootOf(t *testing.T) {
 	repo := &mockAnchorRepo{}
-	anchorID, err := anchors.NewAnchorID(utils.RandomSlice(32))
+	anchorID, err := NewAnchorID(utils.RandomSlice(32))
 	assert.Nil(t, err)
 
-	ethRepo := anchors.NewEthereumAnchorRepository(config.Config, repo)
+	ethRepo := NewEthereumAnchorRepository(config.Config, repo)
 	docRoot := utils.RandomByte32()
 	repo.On("Commits", mock.Anything, mock.Anything).Return(docRoot, nil)
 	gotRoot, err := ethRepo.GetDocumentRootOf(anchorID)

build/configs/default_config.yaml

@@ -32,6 +32,11 @@ networks:
       anchorRepository: "0x7f854dfa98012d7fa55c803bba2260bcdee4b5ed"
       paymentObligation: ""
 
+# Peer-to-peer configurations
+p2p:
+  # Timeout when opening connections to peers
+  connectTimeout: "30s"
+
 # Queue configurations for asynchronous processing
 queue:
   # Defines the number of workers/consumers that will be allocated at startup
@@ -51,6 +56,8 @@ ethereum:
   gasLimit: 4712388
   # Timeout to wait for an ethereum transaction to be added to a block and events triggered
   contextWaitTimeout: "600s"
+  # Timeout to wait for read only operations against ethereum
+  contextReadWaitTimeout: "15s"
   # Node transaction pool max retries to send a transaction over
   maxRetries: 200
   # Node transaction pool interval retry when a concurrent transaction has been detected

config/configuration.go

@@ -44,21 +44,26 @@ type IdentityConfig struct {
 	PrivateKey []byte
 }
 
-// Storage backend
+// GetStoragePath returns the data storage backend
 func (c *Configuration) GetStoragePath() string {
 	return c.V.GetString("storage.Path")
 }
 
-// P2P Port
+// GetP2PPort returns P2P Port
 func (c *Configuration) GetP2PPort() int {
 	return c.V.GetInt("p2p.port")
 }
 
-// P2P External IP
+// GetP2PExternalIP returns P2P External IP
 func (c *Configuration) GetP2PExternalIP() string {
 	return c.V.GetString("p2p.externalIP")
 }
 
+// GetP2PConnectionTimeout returns P2P Connect Timeout
+func (c *Configuration) GetP2PConnectionTimeout() time.Duration {
+	return c.V.GetDuration("p2p.connectTimeout")
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 // Notifications
 ////////////////////////////////////////////////////////////////////////////////
@@ -97,6 +102,10 @@ func (c *Configuration) GetEthereumNodeURL() string {
 	return c.V.GetString("ethereum.nodeURL")
 }
 
+func (c *Configuration) GetEthereumContextReadWaitTimeout() time.Duration {
+	return c.V.GetDuration("ethereum.contextReadWaitTimeout")
+}
+
 func (c *Configuration) GetEthereumContextWaitTimeout() time.Duration {
 	return c.V.GetDuration("ethereum.contextWaitTimeout")
 }

ethereum/geth_client.go

@@ -27,11 +27,23 @@ var log = logging.Logger("geth-client")
 var gc EthereumClient
 var gcInit sync.Once
 
-// GetDefaultContextTimeout retrieves the default duration before an Ethereum call context should time out
+// GetDefaultContextTimeout retrieves the default duration before an Ethereum write call context should time out
 func GetDefaultContextTimeout() time.Duration {
 	return config.Config.GetEthereumContextWaitTimeout()
 }
 
+// GetDefaultReadContextTimeout retrieves the default duration before an Ethereum read call context should time out
+func GetDefaultReadContextTimeout() time.Duration {
+	return config.Config.GetEthereumContextReadWaitTimeout()
+}
+
+// DefaultWaitForReadContext returns context with timeout for read operations
+func DefaultWaitForReadContext() (ctx context.Context, cancelFunc context.CancelFunc) {
+	toBeDone := time.Now().Add(GetDefaultReadContextTimeout())
+	return context.WithDeadline(context.Background(), toBeDone)
+}
+
+// DefaultWaitForTransactionMiningContext returns context with timeout for write operations
 func DefaultWaitForTransactionMiningContext() (ctx context.Context, cancelFunc context.CancelFunc) {
 	toBeDone := time.Now().Add(GetDefaultContextTimeout())
 	return context.WithDeadline(context.TODO(), toBeDone)
@@ -254,8 +266,10 @@ func CalculateIncrement(chainNonce uint64, res map[string]map[string]map[string]
 	}
 }
 
-func GetGethCallOpts() (auth *bind.CallOpts) {
+func GetGethCallOpts() (*bind.CallOpts, context.CancelFunc) {
 	// Assuring that pending transactions are taken into account by go-ethereum when asking for things like
 	// specific transactions and client's nonce
-	return &bind.CallOpts{Pending: true}
+	// with timeout context, in case eth node is not in sync
+	ctx, cancelFunc := DefaultWaitForReadContext()
+	return &bind.CallOpts{Pending: true, Context: ctx}, cancelFunc
 }

identity/ethereum_identity.go

@@ -101,7 +101,8 @@ func (id *EthereumIdentity) FetchKey(key []byte) (Key, error) {
 	if err != nil {
 		return nil, err
 	}
-	opts := ethereum.GetGethCallOpts()
+	// Ignoring cancelFunc as code will block until response or timeout is triggered
+	opts, _ := ethereum.GetGethCallOpts()
 	key32, _ := utils.SliceToByte32(key)
 	keyInstance, err := contract.GetKey(opts, key32)
 	if err != nil {
@@ -135,7 +136,8 @@ func (id *EthereumIdentity) findContract() (exists bool, err error) {
 		return true, nil
 	}
 
-	opts := ethereum.GetGethCallOpts()
+	// Ignoring cancelFunc as code will block until response or timeout is triggered
+	opts, _ := ethereum.GetGethCallOpts()
 	idAddress, err := id.RegistryContract.GetIdentityByCentrifugeId(opts, id.CentrifugeId.BigInt())
 	if err != nil {
 		return false, err
@@ -219,7 +221,8 @@ func (id *EthereumIdentity) fetchKeysByPurpose(keyPurpose int) ([]EthereumIdenti
 	if err != nil {
 		return nil, err
 	}
-	opts := ethereum.GetGethCallOpts()
+	// Ignoring cancelFunc as code will block until response or timeout is triggered
+	opts, _ := ethereum.GetGethCallOpts()
 	bigInt := big.NewInt(int64(keyPurpose))
 	keys, err := contract.GetKeysByPurpose(opts, bigInt)
 	if err != nil {
@@ -372,7 +375,8 @@ func (ids *EthereumIdentityService) CreateIdentity(centrifugeID CentID) (id Iden
 
 // GetIdentityAddress gets the address of the ethereum identity contract for the given CentID
 func (ids *EthereumIdentityService) GetIdentityAddress(centID CentID) (common.Address, error) {
-	opts := ethereum.GetGethCallOpts()
+	// Ignoring cancelFunc as code will block until response or timeout is triggered
+	opts, _ := ethereum.GetGethCallOpts()
 	address, err := ids.registryContract.GetIdentityByCentrifugeId(opts, centID.BigInt())
 	if err != nil {
 		return common.Address{}, err
@@ -389,7 +393,7 @@ func (ids *EthereumIdentityService) LookupIdentityForID(centrifugeID CentID) (Id
 	id := NewEthereumIdentity(centrifugeID, ids.registryContract, ids.config)
 	exists, err := id.CheckIdentityExists()
 	if !exists {
-		return id, fmt.Errorf("identity [%s] does not exist", id.CentrifugeId)
+		return id, fmt.Errorf("identity [%s] does not exist with err [%v]", id.CentrifugeId, err)
 	}
 
 	if err != nil {

p2p/client.go

@@ -17,7 +17,6 @@ import (
 	pstore "github.com/libp2p/go-libp2p-peerstore"
 	ma "github.com/multiformats/go-multiaddr"
 	"google.golang.org/grpc"
-	"google.golang.org/grpc/connectivity"
 )
 
 type Client interface {
@@ -62,17 +61,14 @@ func (d *defaultClient) OpenClient(target string) (p2ppb.P2PServiceClient, error
 	// so LibP2P knows how to contact it
 	hostInstance.Peerstore().AddAddr(peerID, targetAddr, pstore.PermanentAddrTTL)
 
-	// make a new stream from host B to host A
-	g, err := grpcProtoInstance.Dial(context.Background(), peerID, grpc.WithInsecure())
+	// make a new stream from host B to host A with timeout
+	// Retrial is handled internally, connection request will be cancelled by the connection timeout context
+	ctx, _ := context.WithTimeout(context.Background(), config.Config.GetP2PConnectionTimeout())
+	g, err := grpcProtoInstance.Dial(ctx, peerID, grpc.WithInsecure(), grpc.WithBlock())
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to dial peer [%s]: %v", peerID.Pretty(), err)
 	}
 
-	for {
-		if g.GetState() == connectivity.Ready {
-			break
-		}
-	}
 	return p2ppb.NewP2PServiceClient(g), nil
 }
 

p2p/server.go

@@ -203,7 +203,8 @@ func (c *CentP2PServer) makeBasicHost(listenPort int) (host.Host, error) {
 
 	addressFactory := func(addrs []ma.Multiaddr) []ma.Multiaddr {
 		if extMultiAddr != nil {
-			addrs = append(addrs, extMultiAddr)
+			// We currently support a single protocol and transport, if we add more to support then we will need to adapt this code
+			addrs = []ma.Multiaddr{extMultiAddr}
 		}
 		return addrs
 	}

p2p/server_test.go

@@ -73,6 +73,7 @@ func TestCentP2PServer_makeBasicHostWithExternalIP(t *testing.T) {
 	assert.NotNil(t, h)
 	addr, err := ma.NewMultiaddr(fmt.Sprintf("/ip4/%s/tcp/%d", externalIP, listenPort))
 	assert.Nil(t, err)
+	assert.Equal(t, 1, len(h.Addrs()))
 	assert.Contains(t, h.Addrs(), addr)
 }