Merge pull request #1221 from keep-network/channel-channel-channel

BroadcastChannel: improve delivery rate

Two changes to improve the delivery rate of messages.
Create separate broadcast channel for DKG execution

BroadcastChannel libp2p implementation synchronizes on mutex when publishing message to the pubsub and when firing handlers. When we had one channel for the given list of stakers used for all group operations, it could happen that it was used at the same time for DKG and relay entry
signing if the same list of stakers was selected to a new group. As a result, some messages could not reach the destination because mutex synchronization was slowing down message exchange and the given phase could end before all messages were sent and delivered.

This change separates DKG from relay entry signing. We create a temporary channel, just for DKG. The name of the final channel for a group is a group public key in a compressed form and that final channel is used always for every single relay entry signing.
Initialize channel unmarshallers as early as possible

We are sometimes running into a situation when relay entry signing or DKG result publication channel handlers are not yet initialized when the first message is sent by other members.

This change moves initialization of entry signing and DKG result publication protocol handlers a bit earlier.

pkg/beacon/relay/dkg/dkg.go

@@ -27,6 +27,9 @@ func ExecuteDKG(
 	// The staker index should begin with 1
 	playerIndex := group.MemberIndex(index + 1)
 
+	gjkr.RegisterUnmarshallers(channel)
+	dkgResult.RegisterUnmarshallers(channel)
+
 	gjkrResult, gjkrEndBlockHeight, err := gjkr.Execute(
 		playerIndex,
 		groupSize,

pkg/beacon/relay/dkg/result/publish.go

@@ -11,6 +11,16 @@ import (
 	"github.com/keep-network/keep-core/pkg/net"
 )
 
+// RegisterUnmarshallers initializes the given broadcast channel to be able to
+// perform DKG result publication protocol interactions by registering all the
+// required protocol message unmarshallers.
+// The channel needs to be fully initialized before Publish is called.
+func RegisterUnmarshallers(channel net.BroadcastChannel) {
+	channel.RegisterUnmarshaler(func() net.TaggedUnmarshaler {
+		return &DKGResultHashSignatureMessage{}
+	})
+}
+
 // Publish executes Phase 13 and 14 of DKG as a state machine. First, the
 // chosen result is hashed, signed, and sent over a broadcast channel. Then, all
 // other signatures and results are received and accounted for. Those that match
@@ -37,8 +47,6 @@ func Publish(
 		signingStartBlockHeight: startBlockHeight,
 	}
 
-	initializeChannel(channel)
-
 	stateMachine := state.NewMachine(channel, blockCounter, initialState)
 
 	lastState, _, err := stateMachine.Execute(startBlockHeight)
@@ -53,11 +61,3 @@ func Publish(
 
 	return nil
 }
-
-// initializeChannel initializes a given broadcast channel to be able to
-// perform distributed key generation interactions.
-func initializeChannel(channel net.BroadcastChannel) {
-	channel.RegisterUnmarshaler(func() net.TaggedUnmarshaler {
-		return &DKGResultHashSignatureMessage{}
-	})
-}

pkg/beacon/relay/entry/entry.go

@@ -16,7 +16,11 @@ const (
 	signatureBlocks = state.MessagingStateActiveBlocks
 )
 
-func initializeChannel(channel net.BroadcastChannel) {
+// RegisterUnmarshallers initializes the given broadcast channel to be able to
+// perform relay entry signing protocol interactions by registering all the
+// required protocol message unmarshallers.
+// The channel has to be initialized before the SignAndSubmit is called.
+func RegisterUnmarshallers(channel net.BroadcastChannel) {
 	channel.RegisterUnmarshaler(
 		func() net.TaggedUnmarshaler { return &SignatureShareMessage{} })
 }
@@ -33,8 +37,6 @@ func SignAndSubmit(
 	signer *dkg.ThresholdSigner,
 	startBlockHeight uint64,
 ) error {
-	initializeChannel(channel)
-
 	initialState := &signatureShareState{
 		signingStateBase: signingStateBase{
 			channel:         channel,

pkg/beacon/relay/gjkr/gjkr.go

@@ -14,6 +14,34 @@ import (
 
 var logger = log.Logger("keep-gjkr")
 
+// RegisterUnmarshallers initializes the given broadcast channel to be able to
+// perform DKG protocol interactions by registering all the required protocol
+// message unmarshallers.
+// The channel needs to be fully initialized before Execute is called.
+func RegisterUnmarshallers(channel net.BroadcastChannel) {
+	channel.RegisterUnmarshaler(func() net.TaggedUnmarshaler {
+		return &EphemeralPublicKeyMessage{}
+	})
+	channel.RegisterUnmarshaler(func() net.TaggedUnmarshaler {
+		return &MemberCommitmentsMessage{}
+	})
+	channel.RegisterUnmarshaler(func() net.TaggedUnmarshaler {
+		return &PeerSharesMessage{}
+	})
+	channel.RegisterUnmarshaler(func() net.TaggedUnmarshaler {
+		return &SecretSharesAccusationsMessage{}
+	})
+	channel.RegisterUnmarshaler(func() net.TaggedUnmarshaler {
+		return &MemberPublicKeySharePointsMessage{}
+	})
+	channel.RegisterUnmarshaler(func() net.TaggedUnmarshaler {
+		return &PointsAccusationsMessage{}
+	})
+	channel.RegisterUnmarshaler(func() net.TaggedUnmarshaler {
+		return &MisbehavedEphemeralKeysMessage{}
+	})
+}
+
 // Execute runs the GJKR distributed key generation  protocol, given a
 // broadcast channel to mediate with, a block counter used for time tracking,
 // a player index to use in the group, dishonest threshold, and block height
@@ -42,8 +70,6 @@ func Execute(
 		return nil, 0, fmt.Errorf("cannot create a new member: [%v]", err)
 	}
 
-	initializeChannel(channel)
-
 	initialState := &joinState{
 		channel: channel,
 		member:  member,
@@ -63,29 +89,3 @@ func Execute(
 
 	return finalizationState.result(), endBlockHeight, nil
 }
-
-// initializeChannel initializes a given broadcast channel to be able to
-// perform distributed key generation interactions.
-func initializeChannel(channel net.BroadcastChannel) {
-	channel.RegisterUnmarshaler(func() net.TaggedUnmarshaler {
-		return &EphemeralPublicKeyMessage{}
-	})
-	channel.RegisterUnmarshaler(func() net.TaggedUnmarshaler {
-		return &MemberCommitmentsMessage{}
-	})
-	channel.RegisterUnmarshaler(func() net.TaggedUnmarshaler {
-		return &PeerSharesMessage{}
-	})
-	channel.RegisterUnmarshaler(func() net.TaggedUnmarshaler {
-		return &SecretSharesAccusationsMessage{}
-	})
-	channel.RegisterUnmarshaler(func() net.TaggedUnmarshaler {
-		return &MemberPublicKeySharePointsMessage{}
-	})
-	channel.RegisterUnmarshaler(func() net.TaggedUnmarshaler {
-		return &PointsAccusationsMessage{}
-	})
-	channel.RegisterUnmarshaler(func() net.TaggedUnmarshaler {
-		return &MisbehavedEphemeralKeysMessage{}
-	})
-}

pkg/beacon/relay/gjkr/states_test.go

@@ -43,7 +43,7 @@ func TestFullStateTransitions(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		initializeChannel(channel)
+		RegisterUnmarshallers(channel)
 
 		channels[i] = channel
 		states[i] = &joinState{channel, member}

pkg/beacon/relay/node.go

@@ -3,7 +3,6 @@ package relay
 import (
 	"bytes"
 	"crypto/ecdsa"
-	"crypto/sha256"
 	"encoding/hex"
 	"math/big"
 	"sync"
@@ -64,20 +63,11 @@ func (n *Node) JoinGroupIfEligible(
 	}
 
 	if len(indexes) > 0 {
-		// build the channel name and get the broadcast channel
-		broadcastChannelName := channelNameForGroup(groupSelectionResult)
-
-		// We should only join the broadcast channel if we're
-		// elligible for the group
-		broadcastChannel, err := n.netProvider.ChannelFor(
-			broadcastChannelName,
-		)
+		// create temporary broadcast channel for DKG using the group selection
+		// seed
+		broadcastChannel, err := n.netProvider.ChannelFor(newEntry.Text(16))
 		if err != nil {
-			logger.Errorf(
-				"failed to get broadcastChannel for name [%s] with err: [%v]",
-				broadcastChannelName,
-				err,
-			)
+			logger.Errorf("failed to get broadcast channel: [%v]", err)
 			return
 		}
 
@@ -116,10 +106,13 @@ func (n *Node) JoinGroupIfEligible(
 					return
 				}
 
-				err = n.groupRegistry.RegisterGroup(
-					signer,
-					broadcastChannelName,
+				// final broadcast channel name for group is the compressed
+				// public key of the group
+				channelName := hex.EncodeToString(
+					signer.GroupPublicKeyBytesCompressed(),
 				)
+
+				err = n.groupRegistry.RegisterGroup(signer, channelName)
 				if err != nil {
 					logger.Errorf("failed to register a group: [%v]", err)
 				}
@@ -130,22 +123,6 @@ func (n *Node) JoinGroupIfEligible(
 	return
 }
 
-// channelNameForGroup takes the selected stakers, and does the
-// following to construct the broadcastChannel name:
-// * concatenates all of the staker values
-// * returns the hashed concatenated values in hexadecimal representation
-func channelNameForGroup(group *groupselection.Result) string {
-	var channelNameBytes []byte
-	for _, staker := range group.SelectedStakers {
-		channelNameBytes = append(channelNameBytes, staker...)
-	}
-
-	hash := sha256.Sum256(channelNameBytes)
-	hexChannelName := hex.EncodeToString(hash[:])
-
-	return hexChannelName
-}
-
 func candidateGroupMembersFilter(
 	selectedStakers []relaychain.StakerAddress,
 	signing chain.Signing,

pkg/beacon/relay/relay.go

@@ -105,25 +105,23 @@ func (n *Node) GenerateRelayEntry(
 		return
 	}
 
-	for _, signer := range memberships {
-		go func(signer *registry.Membership) {
-			channel, err := n.netProvider.ChannelFor(signer.ChannelName)
-			if err != nil {
-				logger.Errorf(
-					"could not create broadcast channel with name [%v]: [%v]",
-					signer.ChannelName,
-					err,
-				)
-				return
-			}
+	channel, err := n.netProvider.ChannelFor(memberships[0].ChannelName)
+	if err != nil {
+		logger.Errorf("could not create broadcast channel: [%v]", err)
+		return
+	}
+
+	entry.RegisterUnmarshallers(channel)
 
+	for _, member := range memberships {
+		go func(member *registry.Membership) {
 			err = entry.SignAndSubmit(
 				n.blockCounter,
 				channel,
 				relayChain,
 				previousEntry,
 				n.chainConfig.HonestThreshold,
-				signer.Signer,
+				member.Signer,
 				startBlockHeight,
 			)
 			if err != nil {
@@ -133,6 +131,6 @@ func (n *Node) GenerateRelayEntry(
 				)
 				return
 			}
-		}(signer)
+		}(member)
 	}
 }

pkg/internal/dkgtest/dkgtest.go

@@ -14,7 +14,9 @@ import (
 
 	relaychain "github.com/keep-network/keep-core/pkg/beacon/relay/chain"
 	"github.com/keep-network/keep-core/pkg/beacon/relay/dkg"
+	dkgResult "github.com/keep-network/keep-core/pkg/beacon/relay/dkg/result"
 	"github.com/keep-network/keep-core/pkg/beacon/relay/event"
+	"github.com/keep-network/keep-core/pkg/beacon/relay/gjkr"
 	"github.com/keep-network/keep-core/pkg/beacon/relay/group"
 	chainLocal "github.com/keep-network/keep-core/pkg/chain/local"
 	"github.com/keep-network/keep-core/pkg/internal/interception"
@@ -122,6 +124,9 @@ func executeDKG(
 	// make sure all members are up.
 	startBlockHeight := currentBlockHeight + 3
 
+	gjkr.RegisterUnmarshallers(broadcastChannel)
+	dkgResult.RegisterUnmarshallers(broadcastChannel)
+
 	for i := 0; i < relayConfig.GroupSize; i++ {
 		i := i // capture for goroutine
 		go func() {

pkg/internal/entrytest/entrytest.go

@@ -128,6 +128,8 @@ func executeSigning(
 	// make sure all signers are ready
 	startBlockHeight := currentBlockHeight + 3
 
+	entry.RegisterUnmarshallers(broadcastChannel)
+
 	for _, signer := range signers {
 		go func(signer *dkg.ThresholdSigner) {
 			err := entry.SignAndSubmit(