feat: use new chainsyncer

keyper: add contract-addresses to config

keyper-core: implement DB chaincache

gnosis-keyper: use new chainsyncer

gnosis-keyper: remove old sync-handler

gnosisaccessnode: use new chainsyncer

gnosisaccessnode: move storage to package

gnosis-keyper: Rename some functions and types

gnosis-keyper: refactor keyperCore init func

gnosis-keyper: Pass dbpool to keyper core

fixup keyper-core opts

keyper: implement SyncStartBlock / DatabaseChainCache initial syncing

gnosis-keyper: Add int64 overflow checks

gnosis-keyper: add new synchandler

fixup gnosis keyper

chore: avoid var shadow

feat(chainsync): add retry in initial block fetch

chore: rename keypersetmanager synchandler

chore: re-use the gnosis eth-client in keyper-core

chore: refine comments

chore: remove unnecessary import

chore: fix generate, gci, gofumpt

Author: ezdac

rolling-shutter/chainobserver/db/keyper/keyper.sqlc.gen.go

@@ -18,4 +18,42 @@ ORDER BY activation_block_number DESC LIMIT 1;
 
 -- name: GetKeyperSets :many
 SELECT * FROM keyper_set
-ORDER BY activation_block_number ASC;
+ORDER BY activation_block_number ASC;
+
+-- name: InsertSyncedBlock :exec
+INSERT INTO recent_block (
+       block_hash,
+       block_number,
+       parent_hash,
+       timestamp,
+       header
+) VALUES (
+    $1, $2, $3, $4, $5
+) ON CONFLICT DO UPDATE SET
+       block_hash = $1,
+       block_number =$2 ,
+       parent_hash =$3,
+       timestamp =$4 ,
+       header =$5
+;
+
+-- name: GetSyncedBlockByHash :one
+SELECT * FROM recent_block
+WHERE block_hash = $1;
+
+-- name: DeleteSyncedBlockByHash :exec
+DELETE FROM recent_block
+WHERE block_hash = $1;
+
+-- name: GetSyncedBlocks :many
+SELECT * FROM recent_block
+ORDER BY block_number DESC;
+
+-- name: GetLatestSyncedBlocks :many
+SELECT * FROM recent_block
+ORDER BY block_number DESC
+LIMIT $1;
+
+-- name: EvictSyncedBlocksBefore :exec
+DELETE FROM recent_block
+WHERE block_number < $1;

rolling-shutter/chainobserver/db/keyper/sql/queries/keyper.sql

@@ -29,6 +29,8 @@ const (
 )
 
 // EonKeyPublisher is a service that publishes eon keys via a eon key publisher contract.
+// E.g. in the Gnosis keyper implementation this is used to publish keys onchain
+// instead of broadcasting it on the p2p-network.
 type EonKeyPublisher struct {
 	dbpool           *pgxpool.Pool
 	client           *ethclient.Client

rolling-shutter/eonkeypublisher/eonkeypublisher.go

@@ -10,19 +10,20 @@ import (
 
 	"github.com/shutter-network/shutter/shlib/shcrypto"
 
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/gnosisaccessnode/storage"
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/keyperimpl/gnosis"
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/p2pmsg"
 )
 
 type DecryptionKeysHandler struct {
 	config  *Config
-	storage *Storage
+	storage *storage.Memory
 }
 
-func NewDecryptionKeysHandler(config *Config, storage *Storage) *DecryptionKeysHandler {
+func NewDecryptionKeysHandler(config *Config, store *storage.Memory) *DecryptionKeysHandler {
 	return &DecryptionKeysHandler{
 		config:  config,
-		storage: storage,
+		storage: store,
 	}
 }
 

rolling-shutter/gnosisaccessnode/decryptionkeyshandler.go

@@ -2,30 +2,28 @@ package gnosisaccessnode
 
 import (
 	"context"
+	"fmt"
 
+	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/pkg/errors"
-	"github.com/rs/zerolog/log"
 
-	"github.com/shutter-network/shutter/shlib/shcrypto"
-
-	obskeyperdatabase "github.com/shutter-network/rolling-shutter/rolling-shutter/chainobserver/db/keyper"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/gnosisaccessnode/storage"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/gnosisaccessnode/synchandler"
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/chainsync"
-	syncevent "github.com/shutter-network/rolling-shutter/rolling-shutter/medley/chainsync/event"
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/metricsserver"
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/service"
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/p2p"
-	"github.com/shutter-network/rolling-shutter/rolling-shutter/shdb"
 )
 
 type GnosisAccessNode struct {
 	config  *Config
-	storage *Storage
+	storage *storage.Memory
 }
 
 func New(config *Config) *GnosisAccessNode {
 	return &GnosisAccessNode{
 		config:  config,
-		storage: NewStorage(),
+		storage: storage.NewMemory(),
 	}
 }
 
@@ -39,59 +37,39 @@ func (node *GnosisAccessNode) Start(ctx context.Context, runner service.Runner)
 	messageSender.AddMessageHandler(NewDecryptionKeysHandler(node.config, node.storage))
 	services = append(services, messageSender)
 
-	chainSyncClient, err := chainsync.NewClient(
-		ctx,
-		chainsync.WithClientURL(node.config.GnosisNode.EthereumURL),
-		chainsync.WithKeyperSetManager(node.config.Contracts.KeyperSetManager),
-		chainsync.WithKeyBroadcastContract(node.config.Contracts.KeyBroadcastContract),
-		chainsync.WithSyncNewKeyperSet(node.onNewKeyperSet),
-		chainsync.WithSyncNewEonKey(node.onNewEonKey),
+	ethClient, err := ethclient.DialContext(ctx, node.config.GnosisNode.EthereumURL)
+	if err != nil {
+		return err
+	}
+	keyperSetAdded, err := synchandler.NewKeyperSetAdded(
+		ethClient,
+		node.storage,
+		node.config.Contracts.KeyperSetManager,
 	)
 	if err != nil {
-		return errors.Wrap(err, "failed to initialize chain sync client")
+		return err
 	}
-	services = append(services, chainSyncClient)
-
+	eonKeyBroadcast, err := synchandler.NewEonKeyBroadcast(
+		node.storage,
+		node.config.Contracts.KeyBroadcastContract,
+	)
+	if err != nil {
+		return err
+	}
+	chainsyncOpts := []chainsync.Option{
+		chainsync.WithClient(ethClient),
+		chainsync.WithContractEventHandler(keyperSetAdded),
+		chainsync.WithContractEventHandler(eonKeyBroadcast),
+	}
+	chainsyncer, err := chainsync.New(chainsyncOpts...)
+	if err != nil {
+		return fmt.Errorf("can't instantiate chainsync: %w", err)
+	}
+	services = append(services, chainsyncer)
 	if node.config.Metrics.Enabled {
 		metricsServer := metricsserver.New(node.config.Metrics)
 		services = append(services, metricsServer)
 	}
 
 	return runner.StartService(services...)
 }
-
-func (node *GnosisAccessNode) onNewKeyperSet(_ context.Context, keyperSet *syncevent.KeyperSet) error {
-	obsKeyperSet := obskeyperdatabase.KeyperSet{
-		KeyperConfigIndex:     int64(keyperSet.Eon),
-		ActivationBlockNumber: int64(keyperSet.ActivationBlock),
-		Keypers:               shdb.EncodeAddresses(keyperSet.Members),
-		Threshold:             int32(keyperSet.Threshold),
-	}
-	log.Info().
-		Uint64("keyper-config-index", keyperSet.Eon).
-		Uint64("activation-block-number", keyperSet.ActivationBlock).
-		Int("num-keypers", len(keyperSet.Members)).
-		Uint64("threshold", keyperSet.Threshold).
-		Msg("adding keyper set")
-	node.storage.AddKeyperSet(keyperSet.Eon, &obsKeyperSet)
-	return nil
-}
-
-func (node *GnosisAccessNode) onNewEonKey(_ context.Context, eonKey *syncevent.EonPublicKey) error {
-	key := new(shcrypto.EonPublicKey)
-	err := key.Unmarshal(eonKey.Key)
-	if err != nil {
-		log.Error().
-			Err(err).
-			Hex("key", eonKey.Key).
-			Int("keyper-config-index", int(eonKey.Eon)).
-			Msg("received invalid eon key")
-		return nil
-	}
-	log.Info().
-		Int("keyper-config-index", int(eonKey.Eon)).
-		Hex("key", eonKey.Key).
-		Msg("adding eon key")
-	node.storage.AddEonKey(eonKey.Eon, key)
-	return nil
-}