Orchestrator old signatures replay

x/qgb/orchestrator/common_test.go

@@ -2,6 +2,8 @@ package orchestrator
 
 import (
 	"context"
+	tmlog "github.com/tendermint/tendermint/libs/log"
+	"os"
 	"sync"
 	"testing"
 	"time"
@@ -27,6 +29,7 @@ func setupTestOrchestrator(t *testing.T, bc Broadcaster) *orchestrator {
 		orchestratorAddress: crypto.PubkeyToAddress(priv.PublicKey).Hex(),
 		bridgeID:            types.BridgeId,
 		evmPrivateKey:       *priv,
+		logger:              tmlog.NewTMLogger(os.Stdout),
 	}
 }
 

x/qgb/orchestrator/orchestrator.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/ecdsa"
 	"fmt"
+	"github.com/tendermint/tendermint/libs/log"
 	"math/big"
 
 	"github.com/ethereum/go-ethereum/crypto"
@@ -13,6 +14,7 @@ import (
 )
 
 type orchestrator struct {
+	logger log.Logger
 	// TODO this will change once we have the worker pool pattern
 	broadcaster Broadcaster
 
@@ -28,12 +30,14 @@ func (oc *orchestrator) processValsetEvents(ctx context.Context, valsetChannel <
 	for valset := range valsetChannel {
 		signBytes, err := valset.SignBytes(oc.bridgeID)
 		if err != nil {
-			return err
+			oc.logger.Error(fmt.Sprintf("valset nonce %d: %s", valset.Nonce, err.Error()))
+			continue
 		}
 
 		signature, err := types.NewEthereumSignature(signBytes.Bytes(), &oc.evmPrivateKey)
 		if err != nil {
-			return err
+			oc.logger.Error(fmt.Sprintf("valset nonce %d: %s", valset.Nonce, err.Error()))
+			continue
 		}
 
 		// create and send the valset hash
@@ -46,9 +50,10 @@ func (oc *orchestrator) processValsetEvents(ctx context.Context, valsetChannel <
 
 		hash, err := oc.broadcaster.BroadcastTx(ctx, msg)
 		if err != nil {
-			return err
+			oc.logger.Error(fmt.Sprintf("valset nonce %d: %s", valset.Nonce, err.Error()))
+			continue
 		}
-		fmt.Printf("\nsigned Valset %d : %s\n", msg.Nonce, hash)
+		oc.logger.Info(fmt.Sprintf("signed Valset %d : %s", msg.Nonce, hash))
 	}
 	return nil
 }
@@ -61,7 +66,8 @@ func (oc *orchestrator) processDataCommitmentEvents(
 		dataRootHash := types.DataCommitmentTupleRootSignBytes(oc.bridgeID, big.NewInt(int64(dc.Nonce)), dc.Commitment)
 		dcSig, err := types.NewEthereumSignature(dataRootHash.Bytes(), &oc.evmPrivateKey)
 		if err != nil {
-			return err
+			oc.logger.Error(fmt.Sprintf("data commitment range %d-%d: %s", dc.Start, dc.End, err.Error()))
+			continue
 		}
 
 		msg := &types.MsgDataCommitmentConfirm{
@@ -75,9 +81,10 @@ func (oc *orchestrator) processDataCommitmentEvents(
 
 		hash, err := oc.broadcaster.BroadcastTx(ctx, msg)
 		if err != nil {
-			return err
+			oc.logger.Error(fmt.Sprintf("data commitment range %d-%d: %s", dc.Start, dc.End, err.Error()))
+			continue
 		}
-		fmt.Printf("\nsigned commitment %d-%d: %s\n", msg.BeginBlock, msg.EndBlock, hash)
+		oc.logger.Info(fmt.Sprintf("signed commitment %d-%d: %s", msg.BeginBlock, msg.EndBlock, hash))
 	}
 	return nil
 }

x/qgb/orchestrator/orchestrator_client.go

@@ -22,8 +22,13 @@ type orchestratorClient struct {
 	orchestratorAddress string
 }
 
-func NewOrchestratorClient(logger tmlog.Logger, tendermintRpc string, querier Querier, orchAddr string) (AppClient, error) {
-	trpc, err := http.New(tendermintRpc, "/websocket")
+func NewOrchestratorClient(
+	logger tmlog.Logger,
+	tendermintRPC string,
+	querier Querier,
+	orchAddr string,
+) (AppClient, error) {
+	trpc, err := http.New(tendermintRPC, "/websocket")
 	if err != nil {
 		return nil, err
 	}
@@ -41,7 +46,6 @@ func NewOrchestratorClient(logger tmlog.Logger, tendermintRpc string, querier Qu
 	}, nil
 }
 
-// TODO this will be removed when we use the new job/worker design for the client
 func contains(s []uint64, nonce uint64) bool {
 	for _, v := range s {
 		if v == nonce {
@@ -52,17 +56,20 @@ func contains(s []uint64, nonce uint64) bool {
 }
 
 func (oc *orchestratorClient) SubscribeValset(ctx context.Context) (<-chan types.Valset, error) {
-	valsetsChan := make(chan types.Valset, 10)
+	valsetsChan := make(chan types.Valset, 100)
+
+	// will change once we have the new design
+	go oc.addOldValsetAttestations(ctx, valsetsChan) //nolint:errcheck
 
 	results, err := oc.tendermintRPC.Subscribe(
 		ctx,
 		"valset-changes",
 		fmt.Sprintf("%s.%s='%s'", types.EventTypeValsetRequest, sdk.AttributeKeyModule, types.ModuleName),
 	)
-
 	if err != nil {
 		return nil, err
 	}
+
 	nonces := make([]uint64, 10000)
 
 	go func() {
@@ -75,13 +82,13 @@ func (oc *orchestratorClient) SubscribeValset(ctx context.Context) (<-chan types
 				valsets, err := oc.querier.QueryLastValsets(ctx)
 				if err != nil {
 					oc.logger.Error(err.Error())
-					return
+					continue
 				}
 
 				// todo: double check that the first validator set is found
 				if len(valsets) < 1 {
 					oc.logger.Error("no validator sets found")
-					return
+					continue
 				}
 
 				valset := valsets[0]
@@ -90,11 +97,12 @@ func (oc *orchestratorClient) SubscribeValset(ctx context.Context) (<-chan types
 				resp, err := oc.querier.QueryValsetConfirm(ctx, valset.Nonce, oc.orchestratorAddress)
 				if err != nil {
 					oc.logger.Error(err.Error())
-					return
+					continue
 				}
+
 				if resp == nil && !contains(nonces, valset.Nonce) {
-					nonces = append(nonces, valset.Nonce)
 					valsetsChan <- valset
+					nonces = append(nonces, valset.Nonce)
 				}
 			}
 		}
@@ -103,8 +111,66 @@ func (oc *orchestratorClient) SubscribeValset(ctx context.Context) (<-chan types
 	return valsetsChan, nil
 }
 
+func (oc *orchestratorClient) addOldValsetAttestations(ctx context.Context, valsetsChan chan types.Valset) {
+	oc.logger.Info("Started adding Valsets attestation to queue")
+	defer oc.logger.Info("Finished adding Valsets attestation to queue")
+	lastUnbondingHeight, err := oc.querier.QueryLastUnbondingHeight(ctx)
+	if err != nil {
+		oc.logger.Error(err.Error())
+		return
+	}
+
+	valsets, err := oc.querier.QueryLastValsets(ctx)
+	if err != nil {
+		oc.logger.Error(err.Error())
+		return
+	}
+
+	// todo: double check that the first validator set is found
+	if len(valsets) < 1 {
+		oc.logger.Error("no validator sets found")
+		return
+	}
+	valsetsChan <- valsets[0]
+
+	previousNonce := valsets[0].Nonce
+	for {
+		if previousNonce == 1 {
+			return
+		}
+		previousNonce = previousNonce - 1
+		lastVsConfirm, err := oc.querier.QueryValsetConfirm(ctx, previousNonce, oc.orchestratorAddress)
+		if err != nil {
+			oc.logger.Error(err.Error())
+			return
+		}
+		// The valset signed by the orchestrator to get lastVsConfirm
+		// Used to get the height that valset was first introduced
+		correspondingVs, err := oc.querier.QueryValsetByNonce(ctx, previousNonce)
+		if err != nil {
+			oc.logger.Error(err.Error())
+			return
+		}
+		if correspondingVs.Height < lastUnbondingHeight {
+			// Most likely, we're up to date and don't need to catchup anymore
+			return
+		}
+		if lastVsConfirm != nil {
+			// in case we have holes in the signatures
+			continue
+		}
+
+		// valsetChan is the ordinary valset channel used above. The orchestrator keeps adding to it
+		// old attestations same as with new ones when listening.
+		valsetsChan <- *correspondingVs
+	}
+}
+
 func (oc *orchestratorClient) SubscribeDataCommitment(ctx context.Context) (<-chan ExtendedDataCommitment, error) {
-	dataCommitments := make(chan ExtendedDataCommitment)
+	dataCommitments := make(chan ExtendedDataCommitment, 100)
+
+	// will change once we have the new design
+	go oc.addOldDataCommitmentAttestations(ctx, dataCommitments) //nolint:errcheck
 
 	// queryClient := types.NewQueryClient(orchestratorClient.qgbRPC)
 
@@ -164,11 +230,88 @@ func (oc *orchestratorClient) SubscribeDataCommitment(ctx context.Context) (<-ch
 					End:        endHeight,
 					Nonce:      nonce,
 				}
-
 			}
 		}
-
 	}()
 
 	return dataCommitments, nil
 }
+
+func (oc *orchestratorClient) addOldDataCommitmentAttestations(
+	ctx context.Context,
+	dataCommitmentsChan chan ExtendedDataCommitment,
+) {
+	oc.logger.Info("Started adding old Data Commitments attestation to queue")
+	defer oc.logger.Info("Finished adding old Data Commitments attestation to queue")
+	lastUnbondingHeight, err := oc.querier.QueryLastUnbondingHeight(ctx)
+	if err != nil {
+		oc.logger.Error(err.Error())
+		return
+	}
+
+	currentHeight, err := oc.querier.QueryHeight(ctx)
+	if err != nil {
+		oc.logger.Error(err.Error())
+		return
+	}
+
+	var previousBeginBlock uint64
+	var previousEndBlock uint64
+
+	if currentHeight%types.DataCommitmentWindow == 0 {
+		previousBeginBlock = currentHeight
+	} else {
+		// to have a correct range
+		previousBeginBlock = currentHeight - currentHeight%types.DataCommitmentWindow
+	}
+
+	for {
+		// Will be refactored when we have data commitment requests
+		previousEndBlock = previousBeginBlock
+		previousBeginBlock = previousEndBlock - types.DataCommitmentWindow
+
+		if previousEndBlock == 0 {
+			return
+		}
+
+		existingConfirm, err := oc.querier.QueryDataCommitmentConfirm(
+			ctx,
+			previousEndBlock,
+			previousBeginBlock,
+			oc.orchestratorAddress,
+		)
+		if err != nil {
+			oc.logger.Error(err.Error())
+			continue
+		}
+
+		if previousEndBlock < lastUnbondingHeight {
+			// Most likely, we're up to date and don't need to catchup anymore
+			return
+		}
+		if existingConfirm != nil {
+			// In case we have holes in the signatures
+			continue
+		}
+		previousNonce := previousEndBlock / types.DataCommitmentWindow
+
+		previousCommitment, err := oc.tendermintRPC.DataCommitment(
+			ctx,
+			fmt.Sprintf("block.height >= %d AND block.height <= %d",
+				previousBeginBlock,
+				previousEndBlock,
+			),
+		)
+		if err != nil {
+			oc.logger.Error(err.Error())
+			continue
+		}
+
+		dataCommitmentsChan <- ExtendedDataCommitment{
+			Commitment: previousCommitment.DataCommitment,
+			Start:      previousBeginBlock,
+			End:        previousEndBlock,
+			Nonce:      previousNonce,
+		}
+	}
+}

x/qgb/orchestrator/orchestrator_command.go

@@ -67,6 +67,7 @@ func OrchestratorCmd() *cobra.Command {
 				evmPrivateKey:       *config.privateKey,
 				bridgeID:            types.BridgeId,
 				orchestratorAddress: signer.GetSignerInfo().GetAddress().String(),
+				logger:              logger,
 			}
 
 			wg := &sync.WaitGroup{}