feat(chainsync): sync contract event handler with latest head

Author: ezdac

rolling-shutter/keyperimpl/optimism/keyper.go

@@ -144,11 +144,14 @@ func (kpr *Keyper) newEonPublicKey(ctx context.Context, pubKey keyper.EonPublicK
 	log.Info().
 		Uint64("eon", pubKey.Eon).
 		Uint64("activation-block", pubKey.ActivationBlock).
+		Bytes("pub-key", pubKey.PublicKey).
 		Msg("new eon pk")
 	// Currently all keypers call this and race to call this function first.
 	// For now this is fine, but a keyper should only send a transaction if
 	// the key is not set yet.
 	// Best would be a coordinatated leader election who will broadcast the key.
+	// FIXME: the syncer receives an empty key byte.
+	// Is this already
 	tx, err := kpr.l2Client.BroadcastEonKey(ctx, pubKey.Eon, pubKey.PublicKey)
 	if err != nil {
 		log.Error().Err(err).Msg("error broadcasting eon public key")

rolling-shutter/medley/chainsync/client.go

@@ -24,7 +24,7 @@ var noopLogger = &logger.NoopLogger{}
 var ErrServiceNotInstantiated = errors.New("service is not instantiated, pass a handler function option")
 
 type Client struct {
-	client.Client
+	client.EthereumClient
 	log log.Logger
 
 	options *options
@@ -136,7 +136,7 @@ func (s *Client) BroadcastEonKey(ctx context.Context, eon uint64, eonPubKey []by
 // This value is cached, since it is not expected to change.
 func (s *Client) ChainID(ctx context.Context) (*big.Int, error) {
 	if s.chainID == nil {
-		cid, err := s.Client.ChainID(ctx)
+		cid, err := s.EthereumClient.ChainID(ctx)
 		if err != nil {
 			return nil, err
 		}

rolling-shutter/medley/chainsync/client/client.go

@@ -9,7 +9,7 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 )
 
-type Client interface {
+type EthereumClient interface {
 	Close()
 	ChainID(ctx context.Context) (*big.Int, error)
 	BlockByHash(ctx context.Context, hash common.Hash) (*types.Block, error)

rolling-shutter/medley/chainsync/options.go

@@ -24,7 +24,7 @@ type options struct {
 	keyperSetManagerAddress     *common.Address
 	keyBroadcastContractAddress *common.Address
 	clientURL                   string
-	client                      syncclient.Client
+	ethClient                   syncclient.EthereumClient
 	logger                      log.Logger
 	runner                      service.Runner
 	syncStart                   *number.BlockNumber
@@ -37,11 +37,11 @@ type options struct {
 }
 
 func (o *options) verify() error {
-	if o.clientURL != "" && o.client != nil {
+	if o.clientURL != "" && o.ethClient != nil {
 		// TODO: error message
 		return errors.New("can't use client and client url")
 	}
-	if o.clientURL == "" && o.client == nil {
+	if o.clientURL == "" && o.ethClient == nil {
 		// TODO: error message
 		return errors.New("have to provide either url or client")
 	}
@@ -56,24 +56,30 @@ func (o *options) verify() error {
 // of shutter clients background workers.
 func (o *options) apply(ctx context.Context, c *Client) error {
 	var (
-		client syncclient.Client
+		client syncclient.EthereumClient
 		err    error
 	)
 	if o.clientURL != "" {
-		o.client, err = ethclient.DialContext(ctx, o.clientURL)
+		o.ethClient, err = ethclient.DialContext(ctx, o.clientURL)
 		if err != nil {
 			return err
 		}
 	}
-	client = o.client
-	c.log = o.logger
+	client = o.ethClient
 
-	c.Client = client
+	c.EthereumClient = client
 
+	if o.logger != nil {
+		c.log = o.logger
+		// NOCHECKIN:
+		c.log.Info("got logger in options")
+	}
+
+	syncedServices := []syncer.ManualFilterHandler{}
 	// the nil passthrough will use "latest" for each call,
 	// but we want to harmonize and fix the sync start to a specific block.
 	if o.syncStart.IsLatest() {
-		latestBlock, err := c.Client.BlockNumber(ctx)
+		latestBlock, err := c.EthereumClient.BlockNumber(ctx)
 		if err != nil {
 			return errors.Wrap(err, "polling latest block")
 		}
@@ -85,50 +91,69 @@ func (o *options) apply(ctx context.Context, c *Client) error {
 		return err
 	}
 	c.kssync = &syncer.KeyperSetSyncer{
-		Client:     client,
-		Contract:   c.KeyperSetManager,
-		Log:        c.log,
-		StartBlock: o.syncStart,
-		Handler:    o.handlerKeyperSet,
+		Client:              client,
+		Contract:            c.KeyperSetManager,
+		Log:                 c.log,
+		StartBlock:          o.syncStart,
+		Handler:             o.handlerKeyperSet,
+		DisableEventWatcher: true,
 	}
 	if o.handlerKeyperSet != nil {
 		c.services = append(c.services, c.kssync)
+		syncedServices = append(syncedServices, c.kssync)
 	}
 
 	c.KeyBroadcast, err = bindings.NewKeyBroadcastContract(*o.keyBroadcastContractAddress, client)
 	if err != nil {
 		return err
 	}
 	c.epksync = &syncer.EonPubKeySyncer{
-		Client:           client,
-		Log:              c.log,
-		KeyBroadcast:     c.KeyBroadcast,
-		KeyperSetManager: c.KeyperSetManager,
-		Handler:          o.handlerEonPublicKey,
-		StartBlock:       o.syncStart,
+		Client:              client,
+		Log:                 c.log,
+		KeyBroadcast:        c.KeyBroadcast,
+		KeyperSetManager:    c.KeyperSetManager,
+		Handler:             o.handlerEonPublicKey,
+		StartBlock:          o.syncStart,
+		DisableEventWatcher: true,
 	}
 	if o.handlerEonPublicKey != nil {
 		c.services = append(c.services, c.epksync)
+		syncedServices = append(syncedServices, c.epksync)
 	}
 
 	c.sssync = &syncer.ShutterStateSyncer{
-		Client:     client,
-		Contract:   c.KeyperSetManager,
-		Log:        c.log,
-		Handler:    o.handlerShutterState,
-		StartBlock: o.syncStart,
+		Client:              client,
+		Contract:            c.KeyperSetManager,
+		Log:                 c.log,
+		Handler:             o.handlerShutterState,
+		StartBlock:          o.syncStart,
+		DisableEventWatcher: true,
 	}
 	if o.handlerShutterState != nil {
 		c.services = append(c.services, c.sssync)
+		syncedServices = append(syncedServices, c.sssync)
 	}
 
-	if o.handlerBlock != nil {
-		c.uhsync = &syncer.UnsafeHeadSyncer{
-			Client:  client,
-			Log:     c.log,
-			Handler: o.handlerBlock,
+	if o.handlerBlock == nil {
+		// NOOP - but we need to run the UnsafeHeadSyncer.
+		// This is to keep the inner workings consisten,
+		// we use the DisableEventWatcher mechanism in combination
+		// with the UnsafeHeadSyncer instead of the streaming
+		// Watch... subscription on events.
+		// TODO: think about allowing the streaming events,
+		// when guaranteed event order (event1,event2,new-block-event)
+		// is not required
+		o.handlerBlock = func(ctx context.Context, lb *event.LatestBlock) error {
+			return nil
 		}
 	}
+
+	c.uhsync = &syncer.UnsafeHeadSyncer{
+		Client:        client,
+		Log:           c.log,
+		Handler:       o.handlerBlock,
+		SyncedHandler: syncedServices,
+	}
 	if o.handlerBlock != nil {
 		c.services = append(c.services, c.uhsync)
 	}
@@ -141,7 +166,7 @@ func defaultOptions() *options {
 		keyperSetManagerAddress:     &predeploy.KeyperSetManagerAddr,
 		keyBroadcastContractAddress: &predeploy.KeyBroadcastContractAddr,
 		clientURL:                   "",
-		client:                      nil,
+		ethClient:                   nil,
 		logger:                      noopLogger,
 		runner:                      nil,
 		syncStart:                   number.NewBlockNumber(nil),
@@ -193,9 +218,9 @@ func WithLogger(l log.Logger) Option {
 	}
 }
 
-func WithClient(client syncclient.Client) Option {
+func WithClient(client syncclient.EthereumClient) Option {
 	return func(o *options) error {
-		o.client = client
+		o.ethClient = client
 		return nil
 	}
 }

rolling-shutter/medley/chainsync/syncer/eonpubkey.go

@@ -2,11 +2,11 @@ package syncer
 
 import (
 	"context"
-	"errors"
 	"fmt"
 
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/pkg/errors"
 	"github.com/shutter-network/shop-contracts/bindings"
 
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/chainsync/client"
@@ -15,18 +15,54 @@ import (
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/service"
 )
 
+var _ ManualFilterHandler = &EonPubKeySyncer{}
+
 type EonPubKeySyncer struct {
-	Client           client.Client
-	Log              log.Logger
-	KeyBroadcast     *bindings.KeyBroadcastContract
-	KeyperSetManager *bindings.KeyperSetManager
-	StartBlock       *number.BlockNumber
-	Handler          event.EonPublicKeyHandler
+	Client              client.EthereumClient
+	Log                 log.Logger
+	KeyBroadcast        *bindings.KeyBroadcastContract
+	KeyperSetManager    *bindings.KeyperSetManager
+	StartBlock          *number.BlockNumber
+	Handler             event.EonPublicKeyHandler
+	DisableEventWatcher bool
 
 	keyBroadcastCh chan *bindings.KeyBroadcastContractEonKeyBroadcast
 }
 
+func (s *EonPubKeySyncer) QueryAndHandle(ctx context.Context, block uint64) error {
+	s.Log.Info(
+		"pubsyncer query and handle called",
+		"block",
+		block,
+	)
+	opts := &bind.FilterOpts{
+		Start:   block,
+		End:     &block,
+		Context: ctx,
+	}
+	iter, err := s.KeyBroadcast.FilterEonKeyBroadcast(opts)
+	if err != nil {
+		return err
+	}
+	defer iter.Close()
+
+	for iter.Next() {
+		select {
+		case s.keyBroadcastCh <- iter.Event:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+	if err := iter.Error(); err != nil {
+		return errors.Wrap(err, "filter iterator error")
+	}
+	return nil
+}
+
 func (s *EonPubKeySyncer) Start(ctx context.Context, runner service.Runner) error {
+	s.Log.Info(
+		"pubsyncer loop started",
+	)
 	if s.Handler == nil {
 		return errors.New("no handler registered")
 	}
@@ -59,12 +95,14 @@ func (s *EonPubKeySyncer) Start(ctx context.Context, runner service.Runner) erro
 	runner.Defer(func() {
 		close(s.keyBroadcastCh)
 	})
-	subs, err := s.KeyBroadcast.WatchEonKeyBroadcast(watchOpts, s.keyBroadcastCh)
-	// FIXME: what to do on subs.Error()
-	if err != nil {
-		return err
+	if !s.DisableEventWatcher {
+		subs, err := s.KeyBroadcast.WatchEonKeyBroadcast(watchOpts, s.keyBroadcastCh)
+		// FIXME: what to do on subs.Error()
+		if err != nil {
+			return err
+		}
+		runner.Defer(subs.Unsubscribe)
 	}
-	runner.Defer(subs.Unsubscribe)
 	runner.Go(func() error {
 		return s.watchNewEonPubkey(ctx)
 	})
@@ -88,17 +126,23 @@ func (s *EonPubKeySyncer) getInitialPubKeys(ctx context.Context) ([]*event.EonPu
 	if err != nil {
 		return nil, err
 	}
-
 	initialPubKeys := []*event.EonPublicKey{}
+	// NOTE: These are pubkeys that at the state of s.StartBlock
+	// are known to the contracts.
+	// That way we recreate older broadcast publickey events.
+	// We are only interested for keys that belong to keyper-set
+	// that are currently active or will become active in
+	// the future:
 	for i := activeEon; i < numKS; i++ {
 		e, err := s.GetEonPubKeyForEon(ctx, opts, i)
-		// FIXME: translate the error that there is no key
-		// to a continue of the loop
-		// (key not in mapping error, how can we catch that?)
 		if err != nil {
 			return nil, err
 		}
-		initialPubKeys = append(initialPubKeys, e)
+		// if e == nil, this means the keyperset did not broadcast a
+		// key (yet)
+		if e != nil {
+			initialPubKeys = append(initialPubKeys, e)
+		}
 	}
 	return initialPubKeys, nil
 }
@@ -118,11 +162,14 @@ func (s *EonPubKeySyncer) GetEonPubKeyForEon(ctx context.Context, opts *bind.Cal
 		return nil, err
 	}
 	key, err := s.KeyBroadcast.GetEonKey(opts, eon)
-	// XXX: can the key be a null byte?
-	// I think we rather get a index out of bounds error.
 	if err != nil {
 		return nil, err
 	}
+	// NOTE: Solidity returns the null value whenever
+	// one tries to access a key in mapping that doesn't exist
+	if len(key) == 0 {
+		return nil, nil
+	}
 	return &event.EonPublicKey{
 		Eon:           eon,
 		Key:           key,
@@ -137,10 +184,11 @@ func (s *EonPubKeySyncer) watchNewEonPubkey(ctx context.Context) error {
 			if !ok {
 				return nil
 			}
+			pubk := newEonKey.Key
 			bn := newEonKey.Raw.BlockNumber
 			ev := &event.EonPublicKey{
 				Eon:           newEonKey.Eon,
-				Key:           newEonKey.Key,
+				Key:           pubk,
 				AtBlockNumber: number.NewBlockNumber(&bn),
 			}
 			err := s.Handler(ctx, ev)