gnosis-keyper: add synchandler for new chainsyncer

Author: ezdac

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

@@ -33,6 +33,10 @@ SET block_hash = $1, block_number = $2, slot = $3;
 
 -- name: GetTransactionSubmittedEventsSyncedUntil :one
 SELECT * FROM transaction_submitted_events_synced_until LIMIT 1;
+-- name: GetLatestTransactionSubmittedEvent :one
+SELECT * FROM transaction_submitted_event
+ORDER BY block_number DESC
+LIMIT 1;
 
 -- name: GetTransactionSubmittedEventCount :one
 SELECT max(index) + 1 FROM transaction_submitted_event
@@ -41,6 +45,9 @@ WHERE eon = $1;
 -- name: DeleteTransactionSubmittedEventsFromBlockNumber :exec
 DELETE FROM transaction_submitted_event WHERE block_number >= $1;
 
+-- name: DeleteTransactionSubmittedEventsFromBlockHash :exec
+DELETE FROM transaction_submitted_event WHERE block_hash == $1;
+
 -- name: GetTxPointer :one
 SELECT * FROM tx_pointer
 WHERE eon = $1;
@@ -119,4 +126,4 @@ ORDER BY block_number DESC, tx_index DESC, log_index DESC
 LIMIT 1;
 
 -- name: GetNumValidatorRegistrations :one
-SELECT COUNT(*) FROM validator_registrations;
+SELECT COUNT(*) FROM validator_registrations;

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

@@ -0,0 +1,51 @@
+package synchandler
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/hashicorp/go-multierror"
+
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/chainsync/syncer"
+)
+
+var _ syncer.ChainUpdateHandler = &DecryptOnChainUpdate{}
+
+type DecryptionFunction = func(context.Context, *types.Header) error
+
+func NewDecryptOnChainUpdate(fn DecryptionFunction) *DecryptOnChainUpdate {
+	return &DecryptOnChainUpdate{
+		decrypt: fn,
+	}
+}
+
+type DecryptOnChainUpdate struct {
+	decrypt DecryptionFunction
+}
+
+func (cu *DecryptOnChainUpdate) Handle(
+	ctx context.Context,
+	update syncer.ChainUpdateContext,
+) (result error) {
+	// in case of a reorg (non-nil update.Remove segment) we can't roll back any
+	// changes, since the keys have been release already publicly.
+	if update.Append != nil {
+		for _, header := range update.Append.Get() {
+			// We can call the decrypt function with all updated headers,
+			// even if this was a reorg.
+			// This is because the downstream function is expected to keep track of
+			// what slots have already been sent out and decide on itself wether
+			// to re-release keys.
+			err := cu.decrypt(ctx, header)
+			if err != nil {
+				result = multierror.Append(result,
+					fmt.Errorf("failed to decrypt for block %s (num=%d): %w",
+						header.Hash().String(),
+						header.Number.Uint64(),
+						err))
+			}
+		}
+	}
+	return nil
+}

rolling-shutter/keyperimpl/gnosis/synchandler/chainupdate.go

@@ -0,0 +1,160 @@
+package synchandler
+
+import (
+	"context"
+	"fmt"
+	"math"
+
+	"github.com/ethereum/go-ethereum/accounts/abi"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/jackc/pgx/v4"
+	"github.com/jackc/pgx/v4/pgxpool"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog/log"
+	bindings "github.com/shutter-network/gnosh-contracts/gnoshcontracts/sequencer"
+
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/keyperimpl/gnosis/database"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/keyperimpl/gnosis/metrics"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/chainsync/syncer"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/shdb"
+)
+
+func init() {
+	var err error
+	SequencerContractABI, err = bindings.SequencerMetaData.GetAbi()
+	if err != nil {
+		panic(err)
+	}
+}
+
+var SequencerContractABI *abi.ABI
+
+func NewSequencerTransactionSubmitted(dbPool *pgxpool.Pool, address common.Address) (syncer.ContractEventHandler, error) {
+	return syncer.WrapHandler(
+		&SequencerTransactionSubmitted{
+			evABI:   SequencerContractABI,
+			address: address,
+			dbPool:  dbPool,
+		})
+}
+
+type SequencerTransactionSubmitted struct {
+	evABI   *abi.ABI
+	address common.Address
+
+	dbPool *pgxpool.Pool
+}
+
+func (sts *SequencerTransactionSubmitted) Address() common.Address {
+	return sts.address
+}
+
+func (*SequencerTransactionSubmitted) Event() string {
+	return "TransactionSubmitted"
+}
+
+func (sts *SequencerTransactionSubmitted) ABI() abi.ABI {
+	return *sts.evABI
+}
+
+func (sts *SequencerTransactionSubmitted) Accept(
+	_ context.Context,
+	_ types.Header,
+	_ bindings.SequencerTransactionSubmitted,
+) (bool, error) {
+	return true, nil
+}
+
+func (sts *SequencerTransactionSubmitted) Handle(
+	ctx context.Context,
+	update syncer.ChainUpdateContext,
+	events []bindings.SequencerTransactionSubmitted,
+) error {
+	numInsertedEvents := 0
+	numDiscardedEvents := 0
+	err := sts.dbPool.BeginFunc(ctx, func(tx pgx.Tx) error {
+		db := database.New(tx)
+		if update.Remove != nil {
+			for _, header := range update.Remove.Get() {
+				if err := db.DeleteTransactionSubmittedEventsFromBlockHash(ctx, header.Hash().Bytes()); err != nil {
+					return errors.Wrap(err, "failed to delete transaction submitted events from db")
+				}
+			}
+			log.Info().
+				Int("depth", update.Remove.Len()).
+				Int64("previous-synced-until", update.Remove.Latest().Number.Int64()).
+				Int64("new-synced-until", update.Append.Latest().Number.Int64()).
+				Msg("sync status reset due to reorg")
+		}
+		filteredEvents := sts.filterEvents(events)
+		numDiscardedEvents = len(events) - len(filteredEvents)
+		for _, event := range filteredEvents {
+			_, err := db.InsertTransactionSubmittedEvent(ctx, database.InsertTransactionSubmittedEventParams{
+				Index:          int64(event.TxIndex),
+				BlockNumber:    int64(event.Raw.BlockNumber),
+				BlockHash:      event.Raw.BlockHash[:],
+				TxIndex:        int64(event.Raw.TxIndex),
+				LogIndex:       int64(event.Raw.Index),
+				Eon:            int64(event.Eon),
+				IdentityPrefix: event.IdentityPrefix[:],
+				Sender:         shdb.EncodeAddress(event.Sender),
+				GasLimit:       event.GasLimit.Int64(),
+			})
+			if err != nil {
+				return errors.Wrap(err, "failed to insert transaction submitted event into db")
+			}
+			numInsertedEvents++
+			metrics.LatestTxSubmittedEventIndex.WithLabelValues(fmt.Sprint(event.Eon)).Set(float64(event.TxIndex))
+			log.Debug().
+				Uint64("index", event.TxIndex).
+				Uint64("block", event.Raw.BlockNumber).
+				Uint64("eon", event.Eon).
+				Hex("identityPrefix", event.IdentityPrefix[:]).
+				Hex("sender", event.Sender.Bytes()).
+				Uint64("gasLimit", event.GasLimit.Uint64()).
+				Msg("synced new transaction submitted event")
+		}
+		return nil
+	})
+	log.Info().
+		Uint64("start-block", update.Append.Earliest().Number.Uint64()).
+		Uint64("end-block", update.Append.Latest().Number.Uint64()).
+		Int("num-inserted-events", numInsertedEvents).
+		Int("num-discarded-events", numDiscardedEvents).
+		Msg("synced sequencer contract")
+	metrics.TxSubmittedEventsSyncedUntil.Set(float64(update.Append.Latest().Number.Uint64()))
+	return err
+}
+
+func (sts *SequencerTransactionSubmitted) filterEvents(
+	events []bindings.SequencerTransactionSubmitted,
+) []bindings.SequencerTransactionSubmitted {
+	filteredEvents := []bindings.SequencerTransactionSubmitted{}
+	// TODO: before the refactoring, a mux'ed field was read here
+	// in order to filter out Eons that are before "our" first
+	// keyper-sets eon.
+	// Do this by direcly reading "our" keyper-sets from the DB
+	// without manually iterating over all keyper-sets / addresses.
+	// This requires persisting this information upon insertion
+	// in the KeyperSetAdded event handler and thus a database migration.
+	//
+	// For now, insert all events into the DB.
+	// We could clean up this from time to time as a temporary measure
+	// to not make the DB grow unnecessarily.
+	for _, event := range events {
+		if event.Eon > math.MaxInt64 ||
+			!event.GasLimit.IsInt64() {
+			log.Debug().
+				Uint64("eon", event.Eon).
+				Uint64("block-number", event.Raw.BlockNumber).
+				Str("block-hash", event.Raw.BlockHash.Hex()).
+				Uint("tx-index", event.Raw.TxIndex).
+				Uint("log-index", event.Raw.Index).
+				Msg("ignoring transaction submitted event")
+			continue
+		}
+		filteredEvents = append(filteredEvents, event)
+	}
+	return filteredEvents
+}