wip(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/options.go

@@ -70,6 +70,7 @@ func (o *options) apply(ctx context.Context, c *Client) error {
 
 	c.Client = client
 
+	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() {
@@ -85,50 +86,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.kssync)
 	}
 
 	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.kssync)
 	}
 
-	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)
 	}

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

@@ -2,11 +2,12 @@ package syncer
 
 import (
 	"context"
-	"errors"
 	"fmt"
+	"math/big"
 
 	"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,17 +16,45 @@ 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.Client
+	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 {
+	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 {
 	if s.Handler == nil {
 		return errors.New("no handler registered")
@@ -59,12 +88,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)
 	})
@@ -137,10 +168,29 @@ func (s *EonPubKeySyncer) watchNewEonPubkey(ctx context.Context) error {
 			if !ok {
 				return nil
 			}
+			// FIXME: this happens, why?
+			if len(newEonKey.Key) == 0 {
+				opts := &bind.CallOpts{
+					Context:     ctx,
+					BlockNumber: new(big.Int).SetUint64(newEonKey.Raw.BlockNumber),
+				}
+				k, err := s.GetEonPubKeyForEon(ctx, opts, newEonKey.Eon)
+				s.Log.Error(
+					"extra call for GetEonPubKeyForEon errored",
+					"error",
+					err.Error(),
+				)
+				s.Log.Info(
+					"retrieved eon pubkey by getter",
+					"eon",
+					k,
+				)
+			}
+			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)

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

@@ -21,16 +21,52 @@ func makeCallError(attrName string, err error) error {
 
 const channelSize = 10
 
+var _ ManualFilterHandler = &KeyperSetSyncer{}
+
 type KeyperSetSyncer struct {
 	Client     client.Client
 	Contract   *bindings.KeyperSetManager
 	Log        log.Logger
 	StartBlock *number.BlockNumber
 	Handler    event.KeyperSetHandler
+	// disable this when the QueryAndHandle manual polling should be used:
+	DisableEventWatcher bool
 
 	keyperAddedCh chan *bindings.KeyperSetManagerKeyperSetAdded
 }
 
+func (s *KeyperSetSyncer) QueryAndHandle(ctx context.Context, block uint64) error {
+	opts := &bind.FilterOpts{
+		// FIXME: does this work, or do we need index -1 or something?
+		Start:   block,
+		End:     &block,
+		Context: ctx,
+	}
+	iter, err := s.Contract.FilterKeyperSetAdded(opts)
+	// TODO: what errors possible?
+	if err != nil {
+		return err
+	}
+	defer iter.Close()
+
+	for iter.Next() {
+		select {
+		// XXX: this can be nil during the handlers startup.
+		// As far as I understand, a nil channel is never selected.
+		// Will it be selected as soon as the channel is not nil anymore?
+		case s.keyperAddedCh <- iter.Event:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+	// XXX: it looks like this is nil when the iterator is
+	// exhausted without failures
+	if err := iter.Error(); err != nil {
+		return errors.Wrap(err, "filter iterator error")
+	}
+	return nil
+}
+
 func (s *KeyperSetSyncer) Start(ctx context.Context, runner service.Runner) error {
 	if s.Handler == nil {
 		return errors.New("no handler registered")
@@ -69,12 +105,14 @@ func (s *KeyperSetSyncer) Start(ctx context.Context, runner service.Runner) erro
 	runner.Defer(func() {
 		close(s.keyperAddedCh)
 	})
-	subs, err := s.Contract.WatchKeyperSetAdded(watchOpts, s.keyperAddedCh)
-	// FIXME: what to do on subs.Error()
-	if err != nil {
-		return err
+	if !s.DisableEventWatcher {
+		subs, err := s.Contract.WatchKeyperSetAdded(watchOpts, s.keyperAddedCh)
+		// 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.watchNewKeypersService(ctx)
 	})