fix(chainsync): sync all blocks from start to latest head

Before, the chainsync did not fetch all subscribed events
from the provided 'WithSyncStartBlock' to the current latest head.
This is now fixed with a mechanism where the chainsync is progressing
a sync-head in tandem with the incoming updates for the latest head.
The syncer will catch up with polling for all older blocks until the current
latest head is reached.

Author: ezdac

rolling-shutter/medley/chainsync/options.go

@@ -39,132 +39,119 @@ type options struct {
 
 func (o *options) verify() error {
 	if o.clientURL != "" && o.ethClient != nil {
-		// TODO: error message
-		return errors.New("can't use client and client url")
+		return errors.New("'WithClient' and 'WithClientURL' options are mutually exclusive")
 	}
 	if o.clientURL == "" && o.ethClient == nil {
-		// TODO: error message
-		return errors.New("have to provide either url or client")
+		return errors.New("either 'WithClient' or 'WithClientURL' options are expected")
 	}
 	// TODO: check for the existence of the contract addresses depending on
 	// what handlers are not nil
 	return nil
 }
 
-// initialize the shutter client and apply the options.
-// the context is only the initialisation context,
-// and should not be considered to handle the lifecycle
-// of shutter clients background workers.
-func (o *options) apply(ctx context.Context, c *Client) error {
-	var (
-		client syncclient.EthereumClient
-		err    error
-	)
-	if o.clientURL != "" {
-		o.ethClient, err = ethclient.DialContext(ctx, o.clientURL)
-		if err != nil {
-			return err
-		}
-	}
-	client = o.ethClient
-
-	c.EthereumClient = client
-
-	if o.logger != nil {
-		c.log = o.logger
-		// NOCHECKIN:
-		c.log.Info("got logger in options")
-	}
-
+func (o *options) applyHandler(c *Client) error {
+	var err error
 	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.EthereumClient.BlockNumber(ctx)
-		if err != nil {
-			return errors.Wrap(err, "polling latest block")
-		}
-		o.syncStart = number.NewBlockNumber(&latestBlock)
-	}
 
-	c.KeyperSetManager, err = bindings.NewKeyperSetManager(*o.keyperSetManagerAddress, client)
+	c.KeyperSetManager, err = bindings.NewKeyperSetManager(*o.keyperSetManagerAddress, o.ethClient)
 	if err != nil {
 		return err
 	}
 	c.kssync = &syncer.KeyperSetSyncer{
-		Client:                  client,
-		Contract:                c.KeyperSetManager,
-		Log:                     c.log,
-		StartBlock:              o.syncStart,
-		Handler:                 o.handlerKeyperSet,
-		FetchActiveAtStartBlock: o.fetchActivesAtSyncStart,
-		DisableEventWatcher:     true,
+		Client:   o.ethClient,
+		Contract: c.KeyperSetManager,
+		Log:      c.log,
+		Handler:  o.handlerKeyperSet,
 	}
 	if o.handlerKeyperSet != nil {
-		c.services = append(c.services, c.kssync)
 		syncedServices = append(syncedServices, c.kssync)
 	}
 
-	c.KeyBroadcast, err = bindings.NewKeyBroadcastContract(*o.keyBroadcastContractAddress, client)
+	c.KeyBroadcast, err = bindings.NewKeyBroadcastContract(*o.keyBroadcastContractAddress, o.ethClient)
 	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,
-		FetchActiveAtStartBlock: o.fetchActivesAtSyncStart,
-		DisableEventWatcher:     true,
+		Client:           o.ethClient,
+		Log:              c.log,
+		KeyBroadcast:     c.KeyBroadcast,
+		KeyperSetManager: c.KeyperSetManager,
+		Handler:          o.handlerEonPublicKey,
 	}
 	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,
-		FetchActiveAtStartBlock: o.fetchActivesAtSyncStart,
-		DisableEventWatcher:     true,
+		Client:   o.ethClient,
+		Contract: c.KeyperSetManager,
+		Log:      c.log,
+		Handler:  o.handlerShutterState,
 	}
 	if o.handlerShutterState != nil {
-		c.services = append(c.services, c.sssync)
 		syncedServices = append(syncedServices, c.sssync)
 	}
 
 	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
+		// Even if the user is not interested in handling new block events,
+		// the streaming block handler must be running in order to
+		// synchronize polling of new contract events.
+		// Since the handler function is always called, we need to
+		// inject a noop-handler
 		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,
+		Client:             o.ethClient,
+		Log:                c.log,
+		Handler:            o.handlerBlock,
+		SyncedHandler:      syncedServices,
+		FetchActiveAtStart: o.fetchActivesAtSyncStart,
+		SyncStartBlock:     o.syncStart,
 	}
 	if o.handlerBlock != nil {
 		c.services = append(c.services, c.uhsync)
 	}
-	c.privKey = o.privKey
 	return nil
 }
 
+// initialize the shutter client and apply the options.
+// the context is only the initialisation context,
+// and should not be considered to handle the lifecycle
+// of shutter clients background workers.
+func (o *options) apply(ctx context.Context, c *Client) error {
+	var (
+		client syncclient.EthereumClient
+		err    error
+	)
+	if o.clientURL != "" {
+		o.ethClient, err = ethclient.DialContext(ctx, o.clientURL)
+		if err != nil {
+			return err
+		}
+	}
+	client = o.ethClient
+	c.EthereumClient = client
+
+	// 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.EthereumClient.BlockNumber(ctx)
+		if err != nil {
+			return errors.Wrap(err, "polling latest block")
+		}
+		o.syncStart = number.NewBlockNumber(&latestBlock)
+	}
+
+	if o.logger != nil {
+		c.log = o.logger
+	}
+
+	c.privKey = o.privKey
+	return o.applyHandler(c)
+}
+
 func defaultOptions() *options {
 	return &options{
 		keyperSetManagerAddress:     &predeploy.KeyperSetManagerAddr,

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

@@ -12,22 +12,16 @@ import (
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/chainsync/client"
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/chainsync/event"
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/encodeable/number"
-	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/service"
 )
 
 var _ ManualFilterHandler = &EonPubKeySyncer{}
 
 type EonPubKeySyncer struct {
-	Client                  client.EthereumClient
-	Log                     log.Logger
-	KeyBroadcast            *bindings.KeyBroadcastContract
-	KeyperSetManager        *bindings.KeyperSetManager
-	StartBlock              *number.BlockNumber
-	Handler                 event.EonPublicKeyHandler
-	FetchActiveAtStartBlock bool
-	DisableEventWatcher     bool
-
-	keyBroadcastCh chan *bindings.KeyBroadcastContractEonKeyBroadcast
+	Client           client.EthereumClient
+	Log              log.Logger
+	KeyBroadcast     *bindings.KeyBroadcastContract
+	KeyperSetManager *bindings.KeyperSetManager
+	Handler          event.EonPublicKeyHandler
 }
 
 func (s *EonPubKeySyncer) QueryAndHandle(ctx context.Context, block uint64) error {
@@ -48,10 +42,13 @@ func (s *EonPubKeySyncer) QueryAndHandle(ctx context.Context, block uint64) erro
 	defer iter.Close()
 
 	for iter.Next() {
-		select {
-		case s.keyBroadcastCh <- iter.Event:
-		case <-ctx.Done():
-			return ctx.Err()
+		err := s.handle(ctx, iter.Event)
+		if err != nil {
+			s.Log.Error(
+				"handler for `NewKeyperSet` errored",
+				"error",
+				err.Error(),
+			)
 		}
 	}
 	if err := iter.Error(); err != nil {
@@ -60,73 +57,49 @@ func (s *EonPubKeySyncer) QueryAndHandle(ctx context.Context, block uint64) erro
 	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")
-	}
-	// the latest block still has to be fixed.
-	// otherwise we could skip some block events
-	// between the initial poll and the subscription.
-	if s.StartBlock.IsLatest() {
-		latest, err := s.Client.BlockNumber(ctx)
-		if err != nil {
-			return err
-		}
-		s.StartBlock.SetUint64(latest)
-	}
-
-	if s.FetchActiveAtStartBlock {
-		pubKs, err := s.getInitialPubKeys(ctx)
-		if err != nil {
-			return err
-		}
-		for _, k := range pubKs {
-			err := s.Handler(ctx, k)
-			if err != nil {
-				return err
-			}
-		}
+func (s *EonPubKeySyncer) handle(ctx context.Context, newEonKey *bindings.KeyBroadcastContractEonKeyBroadcast) error {
+	pubk := newEonKey.Key
+	bn := newEonKey.Raw.BlockNumber
+	ev := &event.EonPublicKey{
+		Eon:           newEonKey.Eon,
+		Key:           pubk,
+		AtBlockNumber: number.NewBlockNumber(&bn),
+	}
+	err := s.Handler(ctx, ev)
+	if err != nil {
+		return err
 	}
+	return nil
+}
 
-	watchOpts := &bind.WatchOpts{
-		Start:   s.StartBlock.ToUInt64Ptr(),
-		Context: ctx,
+func (s *EonPubKeySyncer) HandleVirtualEvent(ctx context.Context, block *number.BlockNumber) error {
+	pubKs, err := s.getInitialPubKeys(ctx, block)
+	if err != nil {
+		return err
 	}
-	s.keyBroadcastCh = make(chan *bindings.KeyBroadcastContractEonKeyBroadcast, channelSize)
-	runner.Defer(func() {
-		close(s.keyBroadcastCh)
-	})
-	if !s.DisableEventWatcher {
-		subs, err := s.KeyBroadcast.WatchEonKeyBroadcast(watchOpts, s.keyBroadcastCh)
-		// FIXME: what to do on subs.Error()
+	for _, k := range pubKs {
+		err := s.Handler(ctx, k)
 		if err != nil {
 			return err
 		}
-		runner.Defer(subs.Unsubscribe)
 	}
-	runner.Go(func() error {
-		return s.watchNewEonPubkey(ctx)
-	})
 	return nil
 }
 
-func (s *EonPubKeySyncer) getInitialPubKeys(ctx context.Context) ([]*event.EonPublicKey, error) {
+func (s *EonPubKeySyncer) getInitialPubKeys(ctx context.Context, block *number.BlockNumber) ([]*event.EonPublicKey, error) {
 	// This blocknumber specifies AT what state
 	// the contract is called
 	opts := &bind.CallOpts{
 		Context:     ctx,
-		BlockNumber: s.StartBlock.Int,
+		BlockNumber: block.Int,
 	}
 	numKS, err := s.KeyperSetManager.GetNumKeyperSets(opts)
 	if err != nil {
 		return nil, err
 	}
 	// this blocknumber specifies the argument to the contract
 	// getter
-	activeEon, err := s.KeyperSetManager.GetKeyperSetIndexByBlock(opts, s.StartBlock.Uint64())
+	activeEon, err := s.KeyperSetManager.GetKeyperSetIndexByBlock(opts, block.Uint64())
 	if err != nil {
 		return nil, err
 	}
@@ -180,31 +153,3 @@ func (s *EonPubKeySyncer) GetEonPubKeyForEon(ctx context.Context, opts *bind.Cal
 		AtBlockNumber: number.BigToBlockNumber(opts.BlockNumber),
 	}, nil
 }
-
-func (s *EonPubKeySyncer) watchNewEonPubkey(ctx context.Context) error {
-	for {
-		select {
-		case newEonKey, ok := <-s.keyBroadcastCh:
-			if !ok {
-				return nil
-			}
-			pubk := newEonKey.Key
-			bn := newEonKey.Raw.BlockNumber
-			ev := &event.EonPublicKey{
-				Eon:           newEonKey.Eon,
-				Key:           pubk,
-				AtBlockNumber: number.NewBlockNumber(&bn),
-			}
-			err := s.Handler(ctx, ev)
-			if err != nil {
-				s.Log.Error(
-					"handler for `NewKeyperSet` errored",
-					"error",
-					err.Error(),
-				)
-			}
-		case <-ctx.Done():
-			return ctx.Err()
-		}
-	}
-}