Add basic reorg handling

Reorgs are detected by comparing each incoming header with the current
sync head. If the incoming header is supposed to be the child of the
sync head according to its block number, but the parent hash is
different, a reorg is detected.

Author: jannikluhn

rolling-shutter/keyperimpl/gnosis/database/gnosiskeyper.sqlc.gen.go

@@ -38,6 +38,9 @@ SELECT event_count FROM transaction_submitted_event_count
 WHERE eon = $1
 LIMIT 1;
 
+-- name: DeleteTransactionSubmittedEventsFromBlockNumber :exec
+DELETE FROM transaction_submitted_event WHERE block_number >= $1;
+
 -- name: GetTxPointer :one
 SELECT * FROM tx_pointer
 WHERE eon = $1;

rolling-shutter/keyperimpl/gnosis/database/sql/queries/gnosiskeyper.sql

@@ -1,6 +1,7 @@
 package gnosis
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"math"
@@ -20,6 +21,8 @@ import (
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/shdb"
 )
 
+const AssumedReorgDepth = 10
+
 // SequencerSyncer inserts transaction submitted events from the sequencer contract into the database.
 type SequencerSyncer struct {
 	Contract             *sequencerBindings.Sequencer
@@ -30,10 +33,94 @@ type SequencerSyncer struct {
 	SyncStartBlockNumber uint64
 }
 
+// getNumReorgedBlocks returns the number of blocks that have already been synced, but are no
+// longer in the chain.
+func getNumReorgedBlocks(syncedUntil *database.TransactionSubmittedEventsSyncedUntil, header *types.Header) int {
+	shouldBeParent := header.Number.Int64() == syncedUntil.BlockNumber+1
+	isParent := bytes.Equal(header.ParentHash.Bytes(), syncedUntil.BlockHash)
+	isReorg := shouldBeParent && !isParent
+	if !isReorg {
+		return 0
+	}
+	// We don't know how deep the reorg is, so we make a conservative guess. Assuming higher depths
+	// is safer because it means we resync a little bit more.
+	depth := AssumedReorgDepth
+	if syncedUntil.BlockNumber < int64(depth) {
+		return int(syncedUntil.BlockNumber)
+	}
+	return depth
+}
+
+// resetSyncStatus clears the db from its recent history after a reorg of given depth.
+func (s *SequencerSyncer) resetSyncStatus(ctx context.Context, numReorgedBlocks int) error {
+	if numReorgedBlocks == 0 {
+		return nil
+	}
+	return s.DBPool.BeginFunc(ctx, func(tx pgx.Tx) error {
+		queries := database.New(tx)
+
+		syncStatus, err := queries.GetTransactionSubmittedEventsSyncedUntil(ctx)
+		if err != nil {
+			return errors.Wrap(err, "failed to query sync status from db in order to reset it")
+		}
+		if syncStatus.BlockNumber < int64(numReorgedBlocks) {
+			return errors.Wrapf(err, "detected reorg deeper (%d) than blocks synced (%d)", syncStatus.BlockNumber, numReorgedBlocks)
+		}
+
+		deleteFromInclusive := syncStatus.BlockNumber - int64(numReorgedBlocks) + 1
+
+		err = queries.DeleteTransactionSubmittedEventsFromBlockNumber(ctx, deleteFromInclusive)
+		if err != nil {
+			return errors.Wrap(err, "failed to delete transaction submitted events from db")
+		}
+		// Currently, we don't have enough information in the db to populate block hash and slot.
+		// However, using default values here is fine since the syncer is expected to resync
+		// immediately after this function call which will set the correct values. When we do proper
+		// reorg handling, we should store the full block data of the previous blocks so that we can
+		// avoid this.
+		newSyncedUntilBlockNumber := deleteFromInclusive - 1
+		err = queries.SetTransactionSubmittedEventsSyncedUntil(ctx, database.SetTransactionSubmittedEventsSyncedUntilParams{
+			BlockHash:   []byte{},
+			BlockNumber: newSyncedUntilBlockNumber,
+			Slot:        0,
+		})
+		if err != nil {
+			return errors.Wrap(err, "failed to reset transaction submitted event sync status in db")
+		}
+		log.Info().
+			Int("depth", numReorgedBlocks).
+			Int64("previous-synced-until", syncStatus.BlockNumber).
+			Int64("new-synced-until", newSyncedUntilBlockNumber).
+			Msg("sync status reset due to reorg")
+		return nil
+	})
+}
+
+func (s *SequencerSyncer) handlePotentialReorg(ctx context.Context, header *types.Header) error {
+	queries := database.New(s.DBPool)
+	syncedUntil, err := queries.GetTransactionSubmittedEventsSyncedUntil(ctx)
+	if err == pgx.ErrNoRows {
+		return nil
+	}
+	if err != nil {
+		return errors.Wrap(err, "failed to query transaction submitted events sync status")
+	}
+
+	numReorgedBlocks := getNumReorgedBlocks(&syncedUntil, header)
+	if numReorgedBlocks > 0 {
+		return s.resetSyncStatus(ctx, numReorgedBlocks)
+	}
+	return nil
+}
+
 // Sync fetches transaction submitted events from the sequencer contract and inserts them into the
 // database. It starts at the end point of the previous call to sync (or 0 if it is the first call)
 // and ends at the given block number.
 func (s *SequencerSyncer) Sync(ctx context.Context, header *types.Header) error {
+	if err := s.handlePotentialReorg(ctx, header); err != nil {
+		return err
+	}
+
 	queries := database.New(s.DBPool)
 	syncedUntil, err := queries.GetTransactionSubmittedEventsSyncedUntil(ctx)
 	if err != nil && err != pgx.ErrNoRows {