Merge pull request #1277 from keep-network/network-overhaul-5

Context-aware Send BroadcastChannel interface with retransmission support

In #1255 we introduced context-aware message handler in Recv interface of net.BroadcastChannel and in #1260 we enhanced those handlers with retransmission support allowing them to filter out retransmissions of messages that have already been seen. Here, we make Send function context-aware and retransmit the message for the entire lifetime of the
provided context with a frequency provided by retransmission.Ticker set globally.
Retransmissions are filtered out in handlers.

Author: lukasz-zimnoch

cmd/network.go

@@ -18,6 +18,7 @@ import (
 	"github.com/keep-network/keep-core/pkg/net"
 	"github.com/keep-network/keep-core/pkg/net/key"
 	"github.com/keep-network/keep-core/pkg/net/libp2p"
+	"github.com/keep-network/keep-core/pkg/net/retransmission"
 	"github.com/keep-network/keep-core/pkg/operator"
 	"github.com/pborman/uuid"
 	"github.com/urfave/cli"
@@ -92,6 +93,7 @@ func pingRequest(c *cli.Context) error {
 		libp2pConfig,
 		privKey,
 		stakeMonitor,
+		retransmission.NewTimeTicker(ctx, 50*time.Millisecond),
 	)
 	if err != nil {
 		return err
@@ -174,7 +176,7 @@ func pingRequest(c *cli.Context) error {
 
 		go func(msg *PingMessage) {
 			<-start
-			err := broadcastChannel.Send(message)
+			err := broadcastChannel.Send(ctx, message)
 			if err != nil {
 				fmt.Fprintf(
 					os.Stderr,
@@ -209,7 +211,7 @@ func pingRequest(c *cli.Context) error {
 				Sender:  netProvider.ID().String(),
 				Payload: pong + " corresponding to " + pingPayload.Payload,
 			}
-			err := broadcastChannel.Send(message)
+			err := broadcastChannel.Send(ctx, message)
 			if err != nil {
 				return err
 			}

cmd/start.go

@@ -11,6 +11,7 @@ import (
 	"github.com/keep-network/keep-core/pkg/chain/ethereum"
 	"github.com/keep-network/keep-core/pkg/net/key"
 	"github.com/keep-network/keep-core/pkg/net/libp2p"
+	"github.com/keep-network/keep-core/pkg/net/retransmission"
 	"github.com/keep-network/keep-core/pkg/operator"
 	"github.com/urfave/cli"
 )
@@ -64,6 +65,11 @@ func Start(c *cli.Context) error {
 		return fmt.Errorf("error connecting to Ethereum node: [%v]", err)
 	}
 
+	blockCounter, err := chainProvider.BlockCounter()
+	if err != nil {
+		return err
+	}
+
 	stakeMonitor, err := chainProvider.StakeMonitor()
 	if err != nil {
 		return fmt.Errorf("error obtaining stake monitor handle [%v]", err)
@@ -87,6 +93,7 @@ func Start(c *cli.Context) error {
 		config.LibP2P,
 		networkPrivateKey,
 		stakeMonitor,
+		retransmission.NewTicker(blockCounter.WatchBlocks(ctx)),
 	)
 	if err != nil {
 		return err

config.toml.SAMPLE

@@ -25,11 +25,6 @@
 # 	Port = 3920
 #   # Uncomment to override the node's default addresses announced in the network
 #   # AnnouncedAddresses = ["/dns4/example.com/tcp/3919", "/ip4/80.70.60.50/tcp/3919"]
-#
-#   # Number of retransmissions for each message
-#   RetransmissionCycles = 5
-#   # Milliseconds interval between each retransmission of a message
-#   RetransmissionInterval = 1000
 
 [Storage]
   DataDir = "/my/secure/location"

docs/keep-client-quickstart.adoc

@@ -45,18 +45,6 @@ default address then leave `AnnouncedAddresses` set to `[]` or remove the entry
 
 `AnnouncedAddresses = ["/dns4/example.com/tcp/3919", "/ip4/80.70.60.50/tcp/3919"]`
 
-=== Message retransmission
-
-There is no guarantee that all messages sent by the node will reach all of the
-expected receivers. Unexpected network issues may cause messages to be dropped.
-In order to increase delivery rate, message retransmission can be enabled
-by setting `RetransmissionCycles` and `RetransmissionInterval`
-in `[libp2p]` section. The first option determines the number of retransmission
-cycles which should be performed. The second option sets the milliseconds interval
-which should be preserved between cycles. Specific values should be chosen
-reasonably because high number of cycles and short interval value, may lead to
-increased resource usage and network congestion.
-
 == Starting The Client
 
 *Depending on how you orchestrate containers, these steps will vary.  Here we illustrate

infrastructure/kube/templates/keep-client/initcontainer/provision-keep-client/keep-client-config-template.toml

@@ -28,11 +28,5 @@
   # or DNS entries you want to route traffic through.
   AnnouncedAddresses = []
 
-  # TODO: Make these configurable
-  # Number of retransmissions for each message
-  RetransmissionCycles = 5
-  # Milliseconds interval between each retransmission of a message
-  RetransmissionInterval = 1000
-
 [Storage]
   DataDir = "Set by Kube"

infrastructure/kube/templates/keep-client/initcontainer/provision-keep-client/provision-keep-client.js

@@ -220,7 +220,7 @@ async function createKeepClientConfig(operator) {
     Here we format the default rendering to write the config file with Seed/Port values as needed.
     */
     let formattedConfigFile = tomlify.toToml(parsedConfigFile, {
-      replace: (key, value) => { return (key == 'RetransmissionCycles' || key == 'RetransmissionInterval' || key == 'Port') ? value.toFixed(0) : false }
+      replace: (key, value) => { return (key == 'Port') ? value.toFixed(0) : false }
     });
 
     fs.writeFile('/mnt/keep-client/config/keep-client-config.toml', formattedConfigFile, (error) => {

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

@@ -47,7 +47,7 @@ func (rss *resultSigningState) Initiate(ctx context.Context) error {
 	if err != nil {
 		return err
 	}
-	if err := rss.channel.Send(message); err != nil {
+	if err := rss.channel.Send(ctx, message); err != nil {
 		return err
 	}
 	return nil

pkg/beacon/relay/entry/states.go

@@ -69,7 +69,7 @@ func (sss *signatureShareState) Initiate(ctx context.Context) error {
 		sss.MemberIndex(),
 		sss.selfSignatureShare.Marshal(),
 	}
-	if err := sss.channel.Send(message); err != nil {
+	if err := sss.channel.Send(ctx, message); err != nil {
 		return err
 	}
 	return nil

pkg/beacon/relay/gjkr/states.go

@@ -71,7 +71,7 @@ func (ekpgs *ephemeralKeyPairGenerationState) Initiate(ctx context.Context) erro
 		return err
 	}
 
-	if err := ekpgs.channel.Send(message); err != nil {
+	if err := ekpgs.channel.Send(ctx, message); err != nil {
 		return err
 	}
 	return nil
@@ -169,11 +169,11 @@ func (cs *commitmentState) Initiate(ctx context.Context) error {
 		return err
 	}
 
-	if err := cs.channel.Send(sharesMsg); err != nil {
+	if err := cs.channel.Send(ctx, sharesMsg); err != nil {
 		return err
 	}
 
-	if err := cs.channel.Send(commitmentsMsg); err != nil {
+	if err := cs.channel.Send(ctx, commitmentsMsg); err != nil {
 		return err
 	}
 
@@ -251,7 +251,7 @@ func (cvs *commitmentsVerificationState) Initiate(ctx context.Context) error {
 		return err
 	}
 
-	if err := cvs.channel.Send(accusationsMsg); err != nil {
+	if err := cvs.channel.Send(ctx, accusationsMsg); err != nil {
 		return err
 	}
 
@@ -394,7 +394,7 @@ func (pss *pointsShareState) ActiveBlocks() uint64 {
 
 func (pss *pointsShareState) Initiate(ctx context.Context) error {
 	message := pss.member.CalculatePublicKeySharePoints()
-	if err := pss.channel.Send(message); err != nil {
+	if err := pss.channel.Send(ctx, message); err != nil {
 		return err
 	}
 
@@ -457,7 +457,7 @@ func (pvs *pointsValidationState) Initiate(ctx context.Context) error {
 		return err
 	}
 
-	if err := pvs.channel.Send(accusationMsg); err != nil {
+	if err := pvs.channel.Send(ctx, accusationMsg); err != nil {
 		return err
 	}
 
@@ -563,7 +563,7 @@ func (rs *keyRevealState) Initiate(ctx context.Context) error {
 		return err
 	}
 
-	if err := rs.channel.Send(revealMsg); err != nil {
+	if err := rs.channel.Send(ctx, revealMsg); err != nil {
 		return err
 	}
 

pkg/beacon/relay/state/machine_test.go

@@ -30,13 +30,19 @@ func TestExecute(t *testing.T) {
 
 	go func(blockCounter chain.BlockCounter) {
 		blockCounter.WaitForBlockHeight(1)
-		channel.Send(&TestMessage{"message_1"})
+		ctx, cancel := context.WithCancel(context.Background())
+		channel.Send(ctx, &TestMessage{"message_1"})
+		cancel()
 
 		blockCounter.WaitForBlockHeight(4)
-		channel.Send(&TestMessage{"message_2"})
+		ctx, cancel = context.WithCancel(context.Background())
+		channel.Send(ctx, &TestMessage{"message_2"})
+		cancel()
 
 		blockCounter.WaitForBlockHeight(7)
-		channel.Send(&TestMessage{"message_3"})
+		ctx, cancel = context.WithCancel(context.Background())
+		channel.Send(ctx, &TestMessage{"message_3"})
+		cancel()
 	}(blockCounter)
 
 	channel.RegisterUnmarshaler(func() net.TaggedUnmarshaler {