Merge pull request #473 from shutter-network/batch_transition

Avoid crash when sending from inactive keyper

Author: jannikluhn

rolling-shutter/keyper/fx/messagesender.go

@@ -7,6 +7,7 @@ import (
 	"encoding/base64"
 	"encoding/binary"
 	"fmt"
+	"sync/atomic"
 
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog/log"
@@ -31,34 +32,6 @@ type RemoteError struct {
 
 var _ IRetriable = &RemoteError{}
 
-type NonRetriableError struct {
-	Err error
-}
-
-func (*NonRetriableError) IsRetriable() bool {
-	return false
-}
-
-func (e *NonRetriableError) Error() string {
-	return e.Err.Error()
-}
-
-func (e *NonRetriableError) Unwrap() error {
-	return e.Err
-}
-
-var _ IRetriable = &NonRetriableError{}
-
-// IsRetriable checks if we should retry an action that resulted in the given error.
-func IsRetriable(err error) bool {
-	switch e := err.(type) {
-	case IRetriable:
-		return e.IsRetriable()
-	default:
-		return true
-	}
-}
-
 func (remoteError *RemoteError) Error() string {
 	return fmt.Sprintf("remote error: %s", remoteError.msg)
 }
@@ -74,9 +47,10 @@ type MessageSender interface {
 
 // RPCMessageSender signs messages and sends them via RPC to shuttermint.
 type RPCMessageSender struct {
-	rpcclient  client.Client
-	chainID    string
-	signingKey *ecdsa.PrivateKey
+	rpcclient     client.Client
+	chainID       string
+	signingKey    *ecdsa.PrivateKey
+	AllowedToSend *atomic.Bool
 }
 
 var _ MessageSender = &RPCMessageSender{}
@@ -92,15 +66,21 @@ var mockMessageSenderBufferSize = 0x10000
 
 // NewRPCMessageSender creates a new RPCMessageSender.
 func NewRPCMessageSender(cl client.Client, signingKey *ecdsa.PrivateKey) RPCMessageSender {
-	return RPCMessageSender{
+	ms := RPCMessageSender{
 		rpcclient:  cl,
 		chainID:    "",
 		signingKey: signingKey,
 	}
+	ms.AllowedToSend.Store(false)
+	return ms
 }
 
 // SendMessage signs the given shmsg.Message and sends the message to shuttermint.
 func (ms *RPCMessageSender) SendMessage(ctx context.Context, msg *shmsg.Message) error {
+	if !ms.AllowedToSend.Load() {
+		log.Info().Str("msg", msg.String()).Msg("not allowed to send")
+		return nil
+	}
 	if err := ms.maybeFetchChainID(ctx); err != nil {
 		return err
 	}

rolling-shutter/keyper/fx/send.go

@@ -31,7 +31,12 @@ func SendShutterMessages(
 		}
 		err = messageSender.SendMessage(ctx, msg)
 		if err != nil {
-			return err // XXX retry
+			if !isRetrieable(msg) {
+				log.Err(err).Str("msg", msg.String()).Msg("sending non-retrieable msg failed")
+				return err
+			}
+			log.Info().Str("msg", msg.String()).Msg("msg not accepted, will be retried")
+			return nil
 		}
 		log.Info().Int32("id", outgoing.ID).
 			Str("description", outgoing.Description).
@@ -42,3 +47,8 @@ func SendShutterMessages(
 		}
 	}
 }
+
+// FIXME: isRetrieable is a no-op so far.
+func isRetrieable(_ *shmsg.Message) bool {
+	return true
+}

rolling-shutter/keyper/keyper.go

@@ -289,6 +289,9 @@ func (kpr *KeyperCore) handleOnChainKeyperSetChanges(
 		return err
 	}
 	if lastSent == keyperSet.KeyperConfigIndex {
+		log.Debug().
+			Int64("keyper-config-index", keyperSet.KeyperConfigIndex).
+			Msg("batch config already sent (scheduled).")
 		return nil
 	}
 
@@ -309,6 +312,10 @@ func (kpr *KeyperCore) handleOnChainKeyperSetChanges(
 
 	err = q.SetLastBatchConfigSent(ctx, keyperSet.KeyperConfigIndex)
 	if err != nil {
+		log.Warn().Err(err).
+			Interface("keyper-set", keyperSet).
+			Int64("keyper-config-index", keyperSet.KeyperConfigIndex).
+			Msg("error when setting last batch config sent. Returning nil.")
 		return nil
 	}
 
@@ -358,7 +365,9 @@ func (kpr *KeyperCore) operateShuttermint(ctx context.Context, _ service.Runner)
 		if err != nil {
 			return err
 		}
-
+		if !kpr.messageSender.AllowedToSend.Load() {
+			allowSendIfInKeyperSet(ctx, database.New(kpr.dbpool), syncBlockNumber, kpr)
+		}
 		err = fx.SendShutterMessages(ctx, database.New(kpr.dbpool), &kpr.messageSender)
 		if err != nil {
 			return err
@@ -370,3 +379,18 @@ func (kpr *KeyperCore) operateShuttermint(ctx context.Context, _ service.Runner)
 		}
 	}
 }
+
+func allowSendIfInKeyperSet(ctx context.Context, queries *database.Queries, syncBlockNumber uint64, kpr *KeyperCore) {
+	count, err := queries.CountBatchConfigsInBlockRangeWithKeyper(ctx,
+		database.CountBatchConfigsInBlockRangeWithKeyperParams{
+			KeyperAddress: []string{kpr.config.GetAddress().String()},
+			StartBlock:    0,
+			EndBlock:      int64(syncBlockNumber),
+		})
+	if err != nil {
+		log.Err(err).Msg("could not query if in keyper set")
+	}
+	if count > 0 {
+		kpr.messageSender.AllowedToSend.Store(true)
+	}
+}