refactor: organize time-related consts (#60)

* Move stream constants

* Move const WIP

* Move const WIP [2]

* unify usage of "signature received"

* Streamline joinPartyMember

* Streamline cleanupStaleNotifiers

* Fix first tick

* move sig notifier const

* sig notifier ack deadline

* tss keysign/keygen final timeout

* Add comments

* Fix test cases

* Add comments

Author: swift1337

config/const.go

@@ -0,0 +1,82 @@
+// Package config contains the constants for the TSS library
+package config
+
+import (
+	"time"
+
+	"github.com/libp2p/go-libp2p"
+	"github.com/libp2p/go-libp2p/core/protocol"
+	resources "github.com/libp2p/go-libp2p/p2p/host/resource-manager"
+)
+
+const (
+	// StreamTimeoutConnect allow this duration to connect to the peer
+	StreamTimeoutConnect = 10 * time.Second
+
+	// StreamTimeoutRead allow this duration to read from the stream
+	StreamTimeoutRead = 20 * time.Second
+
+	// StreamTimeoutWrite allow this duration to write to the stream
+	StreamTimeoutWrite = 20 * time.Second
+
+	// StreamExcessTTL is the cleanup interval && TTL for "waste" streams (aka streams that are not used)
+	StreamExcessTTL = 1 * time.Minute
+
+	// StreamMaxPayload is the max payload for a stream in bytes. 20MB
+	StreamMaxPayload = 20 << 20
+)
+
+// Signature Notifier related constants
+const (
+	// SigNotifierAckTimeout is the duration to receive ACK back from the peer
+	SigNotifierAckTimeout = 2 * time.Second
+
+	// SigNotifierTTL is the TTL for the notifier. Will be cleaned up if no response from the peer
+	SigNotifierTTL             = 30 * time.Second
+	SigNotifierCleanupInterval = 15 * time.Second
+)
+
+// TSSCommonFinalTimeout is the graceful timeout for keygen/keysign finalization
+const TSSCommonFinalTimeout = 5 * time.Second
+
+// PartyJoinMemberRetryInterval retry interval for joining keygen/keysign party
+const PartyJoinMemberRetryInterval = 500 * time.Millisecond
+
+// ScalingLimits creates a config for libp2p scaling limits
+func ScalingLimits(protocols ...protocol.ID) resources.ConcreteLimitConfig {
+	limits := resources.DefaultLimits
+
+	base := resources.BaseLimit{
+		Streams:         4096,
+		StreamsInbound:  2048,
+		StreamsOutbound: 2048,
+		Memory:          mb(512),
+	}
+
+	increase := resources.BaseLimitIncrease{
+		Streams:         512,
+		StreamsInbound:  256,
+		StreamsOutbound: 256,
+		Memory:          mb(64),
+	}
+
+	limits.ProtocolBaseLimit = base
+	limits.ProtocolLimitIncrease = increase
+
+	for _, protocol := range protocols {
+		limits.AddProtocolLimit(protocol, base, increase)
+		limits.AddProtocolPeerLimit(protocol, base, increase)
+	}
+
+	// Add limits around included libp2p protocols
+	libp2p.SetDefaultServiceLimits(&limits)
+
+	// Turn the scaling limits into a static set of limits using `.AutoScale`. This
+	// scales the limits proportional to system's memory.
+	return limits.AutoScale()
+}
+
+// mb converts a number of megabytes to bytes
+func mb(n int64) int64 {
+	return n << 20
+}

keygen/ecdsa/tss_keygen.go

@@ -15,6 +15,7 @@ import (
 
 	"github.com/zeta-chain/go-tss/blame"
 	"github.com/zeta-chain/go-tss/common"
+	"github.com/zeta-chain/go-tss/config"
 	"github.com/zeta-chain/go-tss/conversion"
 	"github.com/zeta-chain/go-tss/keygen"
 	"github.com/zeta-chain/go-tss/logs"
@@ -151,7 +152,7 @@ func (kg *Keygen) GenerateNewKey(req keygen.Request) (*bcrypto.ECPoint, error) {
 	}
 
 	select {
-	case <-time.After(time.Second * 5):
+	case <-time.After(config.TSSCommonFinalTimeout):
 		close(kg.commStopChan)
 	case <-kg.tssCommonStruct.GetTaskDone():
 		close(kg.commStopChan)

keygen/eddsa/tss_keygen.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/zeta-chain/go-tss/blame"
 	"github.com/zeta-chain/go-tss/common"
+	"github.com/zeta-chain/go-tss/config"
 	"github.com/zeta-chain/go-tss/conversion"
 	"github.com/zeta-chain/go-tss/keygen"
 	"github.com/zeta-chain/go-tss/logs"
@@ -142,9 +143,8 @@ func (kg *Keygen) GenerateNewKey(req keygen.Request) (*bcrypto.ECPoint, error) {
 	}
 
 	select {
-	case <-time.After(time.Second * 5):
+	case <-time.After(config.TSSCommonFinalTimeout):
 		close(kg.commStopChan)
-
 	case <-kg.tssCommonStruct.GetTaskDone():
 		close(kg.commStopChan)
 	}

keysign/ecdsa/tss_keysign.go

@@ -20,6 +20,7 @@ import (
 
 	"github.com/zeta-chain/go-tss/blame"
 	"github.com/zeta-chain/go-tss/common"
+	"github.com/zeta-chain/go-tss/config"
 	"github.com/zeta-chain/go-tss/conversion"
 	"github.com/zeta-chain/go-tss/logs"
 	"github.com/zeta-chain/go-tss/messages"
@@ -207,7 +208,7 @@ func (tKeySign *TssKeySign) SignMessage(
 	}
 
 	select {
-	case <-time.After(time.Second * 5):
+	case <-time.After(config.TSSCommonFinalTimeout):
 		close(tKeySign.commStopChan)
 	case <-tKeySign.tssCommonStruct.GetTaskDone():
 		close(tKeySign.commStopChan)

keysign/eddsa/tss_keysign.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/zeta-chain/go-tss/blame"
 	"github.com/zeta-chain/go-tss/common"
+	"github.com/zeta-chain/go-tss/config"
 	"github.com/zeta-chain/go-tss/conversion"
 	"github.com/zeta-chain/go-tss/logs"
 	"github.com/zeta-chain/go-tss/messages"
@@ -194,7 +195,7 @@ func (tKeySign *KeySign) SignMessage(
 	}
 
 	select {
-	case <-time.After(time.Second * 5):
+	case <-time.After(config.TSSCommonFinalTimeout):
 		close(tKeySign.commStopChan)
 	case <-tKeySign.tssCommonStruct.GetTaskDone():
 		close(tKeySign.commStopChan)

keysign/notifier.go

@@ -13,9 +13,9 @@ import (
 	"github.com/decred/dcrd/dcrec/edwards/v2"
 	"github.com/pkg/errors"
 	"github.com/tendermint/btcd/btcec"
-)
 
-const defaultNotifierTTL = time.Second * 30
+	"github.com/zeta-chain/go-tss/config"
+)
 
 // notifier is design to receive keysign signature, success or failure
 type notifier struct {
@@ -48,7 +48,7 @@ func newNotifier(
 		signatures:  signatures,
 		resp:        make(chan []*common.SignatureData, 1),
 		lastUpdated: time.Now(),
-		ttl:         defaultNotifierTTL,
+		ttl:         config.SigNotifierTTL,
 		lock:        sync.RWMutex{},
 	}, nil
 }

keysign/signature_notifier.go

@@ -13,6 +13,7 @@ import (
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 
+	"github.com/zeta-chain/go-tss/config"
 	"github.com/zeta-chain/go-tss/logs"
 	"github.com/zeta-chain/go-tss/messages"
 	"github.com/zeta-chain/go-tss/p2p"
@@ -52,7 +53,7 @@ func NewSignatureNotifier(host host.Host, logger zerolog.Logger) *SignatureNotif
 		host:         host,
 		notifierLock: &sync.Mutex{},
 		notifiers:    make(map[string]*notifier),
-		streamMgr:    p2p.NewStreamManager(logger),
+		streamMgr:    p2p.NewStreamManager(logger, config.StreamExcessTTL),
 		logger:       logger,
 	}
 
@@ -77,28 +78,37 @@ func (s *SignatureNotifier) Stop() {
 // for longer than notifierTTL. This was added because we allow broadcasts to create notifier objects
 // in handleStream, and need a way to cleanup notifiers that went unused
 func (s *SignatureNotifier) cleanupStaleNotifiers() {
-	doCleanup := func() {
+	cleanup := func() {
 		s.notifierLock.Lock()
-		for messageID, notifier := range s.notifiers {
+		defer s.notifierLock.Unlock()
+
+		var deleted uint
+
+		for msgID, notifier := range s.notifiers {
 			if time.Since(notifier.lastUpdated) > notifier.ttl {
-				delete(s.notifiers, messageID)
+				delete(s.notifiers, msgID)
+				deleted++
 			}
 		}
-		s.notifierLock.Unlock()
+
+		if deleted > 0 {
+			s.logger.Info().Uint("deleted_notifiers", deleted).Msg("Cleaned stale notifiers")
+		}
 	}
-	ticker := time.NewTicker(time.Second * 15)
+
+	ticker := time.NewTicker(config.SigNotifierCleanupInterval)
 	defer ticker.Stop()
 
 	// quickly do an initial cleanup instead of waiting for the ticker. this aids
 	// in testing so we don't have to wait for the ticker to fire
-	doCleanup()
+	cleanup()
 
 	for {
 		select {
 		case <-s.ctx.Done():
 			return
 		case <-ticker.C:
-			doCleanup()
+			cleanup()
 		}
 	}
 }
@@ -149,14 +159,14 @@ func (s *SignatureNotifier) handleStream(stream network.Stream) {
 		signatures = append(signatures, &signature)
 	}
 
-	_, err = s.createOrUpdateNotifier(msg.ID, nil, "", signatures, defaultNotifierTTL)
+	_, err = s.createOrUpdateNotifier(msg.ID, nil, "", signatures, config.SigNotifierTTL)
 	if err != nil {
 		s.logger.Error().Err(err).Msg("fail to update notifier")
 	}
 }
 
 func (s *SignatureNotifier) sendOneMsgToPeer(m *signatureItem) error {
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second*2)
+	ctx, cancel := context.WithTimeout(context.Background(), config.StreamTimeoutConnect)
 	defer cancel()
 
 	stream, err := s.host.NewStream(ctx, m.peerID, p2p.ProtocolSignatureNotifier)
@@ -198,8 +208,8 @@ func (s *SignatureNotifier) sendOneMsgToPeer(m *signatureItem) error {
 		return errors.Wrapf(err, "fail to write KeysignSignature to stream")
 	}
 
-	// we wait for 1 second to allow the receive notify us
-	if err := stream.SetReadDeadline(time.Now().Add(time.Second * 1)); nil != err {
+	// We allow this duration to receive ACK back from the peer
+	if err := p2p.ApplyDeadline(stream, config.SigNotifierAckTimeout, true); err != nil {
 		return errors.Wrapf(err, "fail to set read deadline to stream")
 	}
 

keysign/signature_notifier_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"github.com/zeta-chain/go-tss/config"
 	"github.com/zeta-chain/go-tss/conversion"
 
 	"github.com/zeta-chain/go-tss/common"
@@ -94,58 +95,64 @@ func TestSignatureNotifierHappyPath(t *testing.T) {
 }
 
 func TestSignatureNotifierBroadcastFirst(t *testing.T) {
-	logger := zerolog.Nop()
+	conversion.SetupBech32Prefix()
 
-	poolPubKey := `thorpub1addwnpepq0ul3xt882a6nm6m7uhxj4tk2n82zyu647dyevcs5yumuadn4uamqx7neak`
+	logger := zerolog.New(zerolog.NewConsoleWriter(zerolog.ConsoleTestWriter(t)))
+
+	poolPubKey := "thorpub1addwnpepq0ul3xt882a6nm6m7uhxj4tk2n82zyu647dyevcs5yumuadn4uamqx7neak"
 	messageToSign := "yhEwrxWuNBGnPT/L7PNnVWg7gFWNzCYTV+GuX3tKRH8="
+
 	buf, err := base64.StdEncoding.DecodeString(messageToSign)
 	assert.NoError(t, err)
+
 	messageID, err := common.MsgToHashString(buf)
 	assert.NoError(t, err)
+
 	id1 := tnet.RandIdentityOrFatal(t)
 	id2 := tnet.RandIdentityOrFatal(t)
 	id3 := tnet.RandIdentityOrFatal(t)
+
 	mn := mocknet.New()
-	// add peers to mock net
 
+	// add peers to mock net
 	a1 := tnet.RandLocalTCPAddress()
 	a2 := tnet.RandLocalTCPAddress()
 	a3 := tnet.RandLocalTCPAddress()
 
 	h1, err := mn.AddPeer(id1.PrivateKey(), a1)
-	if err != nil {
-		t.Fatal(err)
-	}
-	p1 := h1.ID()
+	require.NoError(t, err)
+
 	h2, err := mn.AddPeer(id2.PrivateKey(), a2)
-	if err != nil {
-		t.Fatal(err)
-	}
-	p2 := h2.ID()
+	require.NoError(t, err)
+
 	h3, err := mn.AddPeer(id3.PrivateKey(), a3)
-	if err != nil {
-		t.Fatal(err)
-	}
+	require.NoError(t, err)
+
+	p1 := h1.ID()
+	p2 := h2.ID()
 	p3 := h3.ID()
-	if err := mn.LinkAll(); err != nil {
-		t.Error(err)
-	}
-	if err := mn.ConnectAllButSelf(); err != nil {
-		t.Error(err)
-	}
+
+	require.NoError(t, mn.LinkAll())
+	require.NoError(t, mn.ConnectAllButSelf())
+
 	n1 := NewSignatureNotifier(h1, logger)
-	n2 := NewSignatureNotifier(h2, logger)
-	n3 := NewSignatureNotifier(h3, logger)
 	assert.NotNil(t, n1)
+
+	n2 := NewSignatureNotifier(h2, logger)
 	assert.NotNil(t, n2)
+
+	n3 := NewSignatureNotifier(h3, logger)
 	assert.NotNil(t, n3)
+
 	sigFile := "../test_data/signature_notify/sig1.json"
+
 	content, err := os.ReadFile(sigFile)
 	assert.NoError(t, err)
 	assert.NotNil(t, content)
+
 	var signature tsslibcommon.SignatureData
-	err = json.Unmarshal(content, &signature)
-	assert.NoError(t, err)
+	assert.NoError(t, json.Unmarshal(content, &signature))
+
 	sigChan := make(chan string)
 
 	assert.NotContains(t, n1.notifiers, messageID)
@@ -163,11 +170,11 @@ func TestSignatureNotifierBroadcastFirst(t *testing.T) {
 	notifier := n1.notifiers[messageID]
 	n1.notifierLock.Unlock()
 	assert.False(t, notifier.readyToProcess())
-	assert.Equal(t, defaultNotifierTTL, notifier.ttl)
+	assert.Equal(t, config.SigNotifierTTL, notifier.ttl)
 
-	sig, err := n1.WaitForSignature(messageID, [][]byte{buf}, poolPubKey, time.Second*30, sigChan)
-	assert.NoError(t, err)
-	assert.NotNil(t, sig)
+	sig, err := n1.WaitForSignature(messageID, [][]byte{buf}, poolPubKey, config.SigNotifierTTL, sigChan)
+	require.NoError(t, err)
+	require.NotNil(t, sig)
 
 	n1.notifierLock.Lock()
 	assert.NotContains(t, n1.notifiers, messageID)