Introduce broadcast channel retransmission strategies

So far, the broadcast channel used a simple retransmission strategy that
triggered a new retransmission on every new tick. Such an aggressive approach
turned to be not suitable for all use cases. Specifically, this approach tends
to cause network congestion if a big amount of messages is retransmitted for
a long time. To make the retransmission mechanism more flexible, here we
introduce the way to customize it through strategies. We are also add an
implementation of a backoff strategy that decreases its retransmission
frequency over time in an exponential way.

Author: lukasz-zimnoch

pkg/net/libp2p/channel.go

@@ -88,7 +88,11 @@ func (c *channel) Name() string {
 	return c.name
 }
 
-func (c *channel) Send(ctx context.Context, message net.TaggedMarshaler) error {
+func (c *channel) Send(
+	ctx context.Context,
+	message net.TaggedMarshaler,
+	strategy ...net.RetransmissionStrategy,
+) error {
 	messageProto, err := c.messageProto(message)
 	if err != nil {
 		return err
@@ -100,7 +104,21 @@ func (c *channel) Send(ctx context.Context, message net.TaggedMarshaler) error {
 		return c.publish(messageProto)
 	}
 
-	retransmission.ScheduleRetransmissions(ctx, logger, c.retransmissionTicker, doSend)
+	var selectedStrategy net.RetransmissionStrategy
+	switch len(strategy) {
+	case 1:
+		selectedStrategy = strategy[0]
+	default:
+		selectedStrategy = net.StandardRetransmissionStrategy
+	}
+
+	retransmission.ScheduleRetransmissions(
+		ctx,
+		logger,
+		c.retransmissionTicker,
+		doSend,
+		retransmission.WithStrategy(selectedStrategy),
+	)
 
 	return doSend()
 }

pkg/net/local/broadcast_channel.go

@@ -40,7 +40,11 @@ func (lc *localChannel) Name() string {
 	return lc.name
 }
 
-func (lc *localChannel) Send(ctx context.Context, message net.TaggedMarshaler) error {
+func (lc *localChannel) Send(
+	ctx context.Context,
+	message net.TaggedMarshaler,
+	strategy ...net.RetransmissionStrategy,
+) error {
 	bytes, err := message.Marshal()
 	if err != nil {
 		return err
@@ -67,13 +71,22 @@ func (lc *localChannel) Send(ctx context.Context, message net.TaggedMarshaler) e
 		lc.nextSeqno(),
 	)
 
+	var selectedStrategy net.RetransmissionStrategy
+	switch len(strategy) {
+	case 1:
+		selectedStrategy = strategy[0]
+	default:
+		selectedStrategy = net.StandardRetransmissionStrategy
+	}
+
 	retransmission.ScheduleRetransmissions(
 		ctx,
 		logger,
 		lc.retransmissionTicker,
 		func() error {
 			return broadcastMessage(lc.name, netMessage)
 		},
+		retransmission.WithStrategy(selectedStrategy),
 	)
 
 	return broadcastMessage(lc.name, netMessage)

pkg/net/net.go

@@ -2,11 +2,23 @@ package net
 
 import (
 	"context"
-
 	"github.com/keep-network/keep-core/pkg/internal/pb"
 	"github.com/keep-network/keep-core/pkg/operator"
 )
 
+// RetransmissionStrategy represents a specific retransmission strategy.
+type RetransmissionStrategy int
+
+const (
+	// StandardRetransmissionStrategy is the default retransmission strategy
+	// that retransmit the message with a constant frequency.
+	StandardRetransmissionStrategy RetransmissionStrategy = iota
+	// BackoffRetransmissionStrategy is a retransmission strategy that
+	// retransmit the message with an exponentially increasing delay between
+	// subsequent retransmissions.
+	BackoffRetransmissionStrategy
+)
+
 // TransportIdentifier represents a protocol-level identifier. It is an opaque
 // type to the network layer.
 type TransportIdentifier interface {
@@ -93,10 +105,18 @@ type TaggedUnmarshaler interface {
 type BroadcastChannel interface {
 	// Name returns the name of this broadcast channel.
 	Name() string
-	// Send function publishes a message m to the channel. Message m needs to
+	// Send function publishes a message to the channel. Message needs to
 	// conform to the marshalling interface. Message will be periodically
-	// retransmitted by the channel for the lifetime of the provided context.
-	Send(ctx context.Context, m TaggedMarshaler) error
+	// retransmitted by the channel for the lifetime of the provided context
+	// according to the default StandardRetransmissionStrategy. Retransmission
+	// strategy can be set through the `strategy` vararg. If the vararg is
+	// given more than one value, the first value is used as the valid
+	// strategy.
+	Send(
+		ctx context.Context,
+		message TaggedMarshaler,
+		strategy ...RetransmissionStrategy,
+	) error
 	// Recv installs a message handler that will receive messages from the
 	// channel for the entire lifetime of the provided context.
 	// When the context is done, handler is automatically unregistered and

pkg/net/retransmission/retransmission.go

@@ -14,21 +14,25 @@ import (
 	"github.com/keep-network/keep-core/pkg/net"
 )
 
-// ScheduleRetransmissions takes the provided message and retransmits it
-// for every new tick received from the provided Ticker for the entire lifetime
-// of the Context calling the provided retransmit function. The retransmit
-// function has to guarantee that every call from this function sends a message
-// with the same sequence number.
+// RetransmitFn represents a retransmission routine.
+type RetransmitFn func() error
+
+// ScheduleRetransmissions uses the given Strategy to decide whether to call
+// the provided RetransmitFn for every new tick received from the provided
+// Ticker for the entire lifetime of the Context. The RetransmitFn function has
+// to guarantee that every call from this function sends a message with the
+// same sequence number.
 func ScheduleRetransmissions(
 	ctx context.Context,
 	logger log.StandardLogger,
 	ticker *Ticker,
-	retransmit func() error,
+	retransmit RetransmitFn,
+	strategy Strategy,
 ) {
 	go func() {
 		ticker.onTick(ctx, func() {
 			go func() {
-				if err := retransmit(); err != nil {
+				if err := strategy.Tick(retransmit); err != nil {
 					logger.Errorf("could not retransmit message: [%v]", err)
 				}
 			}()

pkg/net/retransmission/retransmission_test.go

@@ -23,6 +23,7 @@ func TestRetransmitExpectedNumberOfTimes(t *testing.T) {
 			atomic.AddUint64(&retransmissionsCount, 1)
 			return nil
 		},
+		WithStandardStrategy(),
 	)
 
 	<-ctx.Done()

pkg/net/retransmission/strategy.go

@@ -0,0 +1,74 @@
+package retransmission
+
+import "github.com/keep-network/keep-core/pkg/net"
+
+// Strategy represents a specific retransmission strategy.
+type Strategy interface {
+	// Tick asks the strategy to run the provided retransmission routine.
+	// The strategy uses their internal state and logic to decide whether to
+	// call the retransmission function or not.
+	Tick(retransmitFn RetransmitFn) error
+}
+
+// WithStrategy is a strategy factory function that returns the requested
+// strategy instance.
+func WithStrategy(strategy net.RetransmissionStrategy) Strategy {
+	switch strategy {
+	case net.StandardRetransmissionStrategy:
+		return WithStandardStrategy()
+	case net.BackoffRetransmissionStrategy:
+		return WithBackoffStrategy()
+	default:
+		panic("retransmission strategy not implemented")
+	}
+}
+
+// StandardStrategy is the basic retransmission strategy that triggers the
+// retransmission routine on every tick.
+type StandardStrategy struct{}
+
+// WithStandardStrategy uses the StandardStrategy as the retransmission
+// strategy.
+func WithStandardStrategy() *StandardStrategy {
+	return &StandardStrategy{}
+}
+
+// Tick implements the Strategy.Tick function.
+func (ss *StandardStrategy) Tick(retransmitFn RetransmitFn) error {
+	return retransmitFn()
+}
+
+// BackoffStrategy is a retransmission strategy that triggers the retransmission
+// routine with an exponentially increasing delay. That is, the delay between
+// first and second retransmission is 1 tick, between second and third is 2
+// ticks, between third and fourth is 4 ticks and so on. Graphically, the
+// schedule looks as follows: R _ R _ _ R _ _ _ _  R _ _ _ _ _ _ _ _ R
+type BackoffStrategy struct {
+	tickCounter    uint64
+	delay          uint64
+	retransmitTick uint64
+}
+
+// WithBackoffStrategy uses the BackoffStrategy as the retransmission
+// strategy.
+func WithBackoffStrategy() *BackoffStrategy {
+	return &BackoffStrategy{
+		tickCounter:    0,
+		delay:          1,
+		retransmitTick: 1,
+	}
+}
+
+// Tick implements the Strategy.Tick function.
+func (bos *BackoffStrategy) Tick(retransmitFn RetransmitFn) error {
+	bos.tickCounter++
+
+	if bos.tickCounter == bos.retransmitTick {
+		bos.retransmitTick += bos.delay + 1
+		bos.delay *= 2
+
+		return retransmitFn()
+	}
+
+	return nil
+}

pkg/net/retransmission/strategy_test.go

@@ -0,0 +1,79 @@
+package retransmission
+
+import (
+	"reflect"
+	"testing"
+)
+
+func TestStandardStrategy(t *testing.T) {
+	strategy := WithStandardStrategy()
+
+	retransmitInvocations := make(map[int]bool)
+
+	for i := 1; i <= 10; i++ {
+		err := strategy.Tick(func() error {
+			retransmitInvocations[i] = true
+			return nil
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	expectedRetransmitInvocations := map[int]bool{
+		1:  true,
+		2:  true,
+		3:  true,
+		4:  true,
+		5:  true,
+		6:  true,
+		7:  true,
+		8:  true,
+		9:  true,
+		10: true,
+	}
+	if !reflect.DeepEqual(expectedRetransmitInvocations, retransmitInvocations) {
+		t.Errorf(
+			"unexpected invocations\n"+
+				"expected: [%v]\n"+
+				"actual:   [%v]",
+			expectedRetransmitInvocations,
+			retransmitInvocations,
+		)
+	}
+}
+
+func TestBackoffStrategy(t *testing.T) {
+	strategy := WithBackoffStrategy()
+
+	retransmitInvocations := make(map[int]bool)
+
+	for i := 1; i <= 100; i++ {
+		err := strategy.Tick(func() error {
+			retransmitInvocations[i] = true
+			return nil
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	expectedRetransmitInvocations := map[int]bool{
+		1:  true,
+		3:  true,
+		6:  true,
+		11: true,
+		20: true,
+		37: true,
+		70: true,
+	}
+	if !reflect.DeepEqual(expectedRetransmitInvocations, retransmitInvocations) {
+		t.Errorf(
+			"unexpected invocations\n"+
+				"expected: [%v]\n"+
+				"actual:   [%v]",
+			expectedRetransmitInvocations,
+			retransmitInvocations,
+		)
+	}
+}