chore(op): improve sync client

Author: ezdac

rolling-shutter/keyperimpl/optimism/bootstrap/config.go

@@ -19,7 +19,7 @@ func NewConfig() *Config {
 
 func (c *Config) Init() {
 	c.SigningKey = &keys.ECDSAPrivate{}
-	c.ByActivationBlockNumber = number.NewBlockNumber()
+	c.ByActivationBlockNumber = number.NewBlockNumber(nil)
 }
 
 type Config struct {

rolling-shutter/keyperimpl/optimism/keyper.go

@@ -3,6 +3,7 @@ package optimism
 import (
 	"context"
 
+	"github.com/ethereum/go-ethereum/accounts/abi/bind"
 	"github.com/jackc/pgx/v4"
 	"github.com/jackc/pgx/v4/pgxpool"
 	"github.com/pkg/errors"
@@ -28,9 +29,10 @@ import (
 var ErrParseKeyperSet = errors.New("can't parse KeyperSet")
 
 type Keyper struct {
-	core   *keyper.KeyperCore
-	dbpool *pgxpool.Pool
-	config *config.Config
+	core     *keyper.KeyperCore
+	l2Client *shopclient.ShutterL2Client
+	dbpool   *pgxpool.Pool
+	config   *config.Config
 
 	trigger chan<- *broker.Event[*epochkghandler.DecryptionTrigger]
 }
@@ -50,7 +52,6 @@ func (kpr *Keyper) Start(ctx context.Context, runner service.Runner) error {
 	}
 	kpr.dbpool = dbpool
 
-	// TODO: the new latest block handler function will put values into this channel
 	trigger := make(chan *broker.Event[*epochkghandler.DecryptionTrigger])
 	kpr.trigger = trigger
 
@@ -77,17 +78,17 @@ func (kpr *Keyper) Start(ctx context.Context, runner service.Runner) error {
 		return errors.Wrap(err, "can't instantiate keyper core")
 	}
 	// TODO: wrap the logger and pass in
-	l2Client, err := shopclient.NewShutterL2Client(
+	kpr.l2Client, err = shopclient.NewShutterL2Client(
 		ctx,
 		shopclient.WithClientURL(kpr.config.Optimism.JSONRPCURL),
 		shopclient.WithSyncNewBlock(kpr.newBlock),
 		shopclient.WithSyncNewKeyperSet(kpr.newKeyperSet),
+		shopclient.WithPrivateKey(kpr.config.Optimism.PrivateKey.Key),
 	)
-	// TODO: how to deal with polling past state? (sounds like a big addition to the l2Client)
 	if err != nil {
 		return err
 	}
-	return runner.StartService(kpr.core, l2Client)
+	return runner.StartService(kpr.core, kpr.l2Client)
 }
 
 func (kpr *Keyper) newBlock(_ context.Context, ev *shopevent.LatestBlock) error {
@@ -139,11 +140,34 @@ func (kpr *Keyper) newKeyperSet(ctx context.Context, ev *shopevent.KeyperSet) er
 	})
 }
 
-func (kpr *Keyper) newEonPublicKey(ctx context.Context, pk keyper.EonPublicKey) error {
+func (kpr *Keyper) newEonPublicKey(ctx context.Context, pubKey keyper.EonPublicKey) error {
 	log.Info().
-		Uint64("eon", pk.Eon).
-		Uint64("activation-block", pk.ActivationBlock).
+		Uint64("eon", pubKey.Eon).
+		Uint64("activation-block", pubKey.ActivationBlock).
 		Msg("new eon pk")
-	// TODO: post the public key to the contract
+	// 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.
+	tx, err := kpr.l2Client.BroadcastEonKey(ctx, pubKey.Eon, pubKey.PublicKey)
+	if err != nil {
+		log.Error().Err(err).Msg("error broadcasting eon public key")
+		return errors.Wrap(err, "error broadcasting eon public-key")
+	}
+	log.Info().
+		Str("hash", tx.Hash().Hex()).
+		Msg("sent eon pubkey transaction")
+
+	receipt, err := bind.WaitMined(ctx, kpr.l2Client, tx)
+	if err != nil {
+		log.Error().Err(err).Msg("error waiting for pubkey tx mined")
+		return err
+	}
+	// NOCHECKIN: log the JSON receipt or only specific fields
+	log.Info().
+		Interface("receipt", receipt).
+		Msg("eon pubkey transaction mined")
+	// TODO:
+	// wait / confirm of tx, otherwise resend
 	return nil
 }

rolling-shutter/keyperimpl/optimism/sync/client.go

@@ -2,9 +2,12 @@ package sync
 
 import (
 	"context"
+	"crypto/ecdsa"
 	"io"
+	"math/big"
 
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
+	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/pkg/errors"
 	"github.com/shutter-network/shop-contracts/bindings"
@@ -32,6 +35,8 @@ type ShutterL2Client struct {
 	log log.Logger
 
 	options *options
+	chainID *big.Int
+	privKey *ecdsa.PrivateKey
 
 	KeyperSetManager *bindings.KeyperSetManager
 	KeyBroadcast     *bindings.KeyBroadcastContract
@@ -74,26 +79,6 @@ func (s *ShutterL2Client) getServices() []service.Service {
 	return s.services
 }
 
-//	func syncInitial() {
-//		if s.ForceEmitActiveKeyperSet {
-//			var b *number.BlockNumber
-//			if s.StartBlock == nil {
-//				b = number.LatestBlock
-//			} else {
-//				b = number.NewBlockNumber()
-//				b.SetInt64(int64(*s.StartBlock))
-//			}
-//			activeKSAddred, err := s.getKeyperSetForBlock(ctx, b)
-//			if err != nil {
-//				return err
-//			}
-//			err = s.handler(activeKSAddred)
-//			if err != nil {
-//				return errors.Wrap(err, "handling of forced emission of active keyper set failed")
-//			}
-//		}
-//	}
-
 func (s *ShutterL2Client) GetShutterState(ctx context.Context) (*event.ShutterState, error) {
 	if s.sssync == nil {
 		return nil, errors.Wrap(ErrServiceNotInstantiated, "ShutterStateSyncer service not instantiated")
@@ -126,14 +111,46 @@ func (s *ShutterL2Client) GetKeyperSetForBlock(ctx context.Context, b *number.Bl
 
 func (s *ShutterL2Client) GetEonPubKeyForEon(ctx context.Context, eon uint64) (*event.EonPublicKey, error) {
 	if s.sssync == nil {
-		return nil, errors.Wrap(ErrServiceNotInstantiated, "EonPubKeySyncer service not instantiated")
+		return nil, errors.Wrap(ErrServiceNotInstantiated, "ShutterStateSyncer service not instantiated")
 	}
 	opts := &bind.CallOpts{
 		Context: ctx,
 	}
 	return s.epksync.GetEonPubKeyForEon(ctx, opts, eon)
 }
 
-func (s *ShutterL2Client) Start(ctx context.Context, runner service.Runner) error {
+func (s *ShutterL2Client) BroadcastEonKey(ctx context.Context, eon uint64, eonPubKey []byte) (*types.Transaction, error) {
+	// TODO: first do a getEonKey. If we already have something (ideally the same)
+	// don't do a transaction
+	// s.KeyBroadcast.GetEonKey(eon)
+	if s.privKey == nil {
+		return nil, errors.New("can't broadcast eon public-key, client does not have a signer set")
+	}
+	chainID, err := s.ChainID(ctx)
+	if err != nil {
+		return nil, errors.Wrap(err, "retrieve chain id")
+	}
+	opts, err := bind.NewKeyedTransactorWithChainID(s.privKey, chainID)
+	if err != nil {
+		return nil, errors.Wrap(err, "construct signer transaction opts")
+	}
+	opts.Context = ctx
+	return s.KeyBroadcast.BroadcastEonKey(opts, eon, eonPubKey)
+}
+
+// ChainID returns the chainid of the underlying L2 chain.
+// This value is cached, since it is not expected to change.
+func (s *ShutterL2Client) ChainID(ctx context.Context) (*big.Int, error) {
+	if s.chainID == nil {
+		cid, err := s.Client.ChainID(ctx)
+		if err != nil {
+			return nil, err
+		}
+		s.chainID = cid
+	}
+	return s.chainID, nil
+}
+
+func (s *ShutterL2Client) Start(_ context.Context, runner service.Runner) error {
 	return runner.StartService(s.getServices()...)
 }

rolling-shutter/keyperimpl/optimism/sync/options.go

@@ -2,17 +2,19 @@ package sync
 
 import (
 	"context"
-	"errors"
+	"crypto/ecdsa"
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/ethclient"
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/pkg/errors"
 	"github.com/shutter-network/shop-contracts/bindings"
 	"github.com/shutter-network/shop-contracts/predeploy"
 
 	syncclient "github.com/shutter-network/rolling-shutter/rolling-shutter/keyperimpl/optimism/sync/client"
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/keyperimpl/optimism/sync/event"
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/keyperimpl/optimism/sync/syncer"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/encodeable/number"
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/service"
 )
 
@@ -25,7 +27,8 @@ type options struct {
 	client                      syncclient.Client
 	logger                      log.Logger
 	runner                      service.Runner
-	syncStart                   *uint64
+	syncStart                   *number.BlockNumber
+	privKey                     *ecdsa.PrivateKey
 
 	handlerShutterState event.ShutterStateHandler
 	handlerKeyperSet    event.KeyperSetHandler
@@ -67,6 +70,16 @@ func (o *options) apply(ctx context.Context, c *ShutterL2Client) error {
 
 	c.Client = 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.Client.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)
 	if err != nil {
 		return err
@@ -118,6 +131,7 @@ func (o *options) apply(ctx context.Context, c *ShutterL2Client) error {
 	if o.handlerBlock != nil {
 		c.services = append(c.services, c.uhsync)
 	}
+	c.privKey = o.privKey
 	return nil
 }
 
@@ -129,14 +143,16 @@ func defaultOptions() *options {
 		client:                      nil,
 		logger:                      noopLogger,
 		runner:                      nil,
-		syncStart:                   nil,
+		syncStart:                   number.NewBlockNumber(nil),
 	}
 }
 
-func WithSyncStartBlock(blockNumber uint64) Option {
+func WithSyncStartBlock(blockNumber *number.BlockNumber) Option {
+	if blockNumber == nil {
+		blockNumber = number.NewBlockNumber(nil)
+	}
 	return func(o *options) error {
-		bn := blockNumber
-		o.syncStart = &bn
+		o.syncStart = blockNumber
 		return nil
 	}
 }
@@ -183,6 +199,13 @@ func WithClient(client syncclient.Client) Option {
 	}
 }
 
+func WithPrivateKey(key *ecdsa.PrivateKey) Option {
+	return func(o *options) error {
+		o.privKey = key
+		return nil
+	}
+}
+
 func WithSyncNewKeyperSet(handler event.KeyperSetHandler) Option {
 	return func(o *options) error {
 		o.handlerKeyperSet = handler

rolling-shutter/keyperimpl/optimism/sync/syncer/eonpubkey.go

@@ -11,43 +11,99 @@ import (
 
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/keyperimpl/optimism/sync/client"
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/keyperimpl/optimism/sync/event"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/encodeable/number"
 	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/service"
 )
 
 type EonPubKeySyncer struct {
 	Client     client.Client
 	Log        log.Logger
 	Contract   *bindings.KeyBroadcastContract
-	StartBlock *uint64
+	StartBlock *number.BlockNumber
 	Handler    event.EonPublicKeyHandler
 
 	keyBroadcastCh chan *bindings.KeyBroadcastContractEonKeyBroadcast
+	ksManager      *bindings.KeyperSetManager
 }
 
 func (s *EonPubKeySyncer) Start(ctx context.Context, runner service.Runner) error {
 	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)
+	}
+	pubKs, err := s.getInitialPubKeys(ctx)
+	if err != nil {
+		return err
+	}
+	for _, k := range pubKs {
+		err := s.Handler(ctx, k)
+		if err != nil {
+			return err
+		}
+	}
+
 	watchOpts := &bind.WatchOpts{
-		Start:   s.StartBlock, // nil means latest
+		Start:   s.StartBlock.ToUInt64Ptr(),
 		Context: ctx,
 	}
-	s.keyBroadcastCh = make(chan *bindings.KeyBroadcastContractEonKeyBroadcast, 10)
+	s.keyBroadcastCh = make(chan *bindings.KeyBroadcastContractEonKeyBroadcast, channelSize)
+	runner.Defer(func() {
+		close(s.keyBroadcastCh)
+	})
 	subs, err := s.Contract.WatchEonKeyBroadcast(watchOpts, s.keyBroadcastCh)
 	// FIXME: what to do on subs.Error()
 	if err != nil {
 		return err
 	}
 	runner.Defer(subs.Unsubscribe)
-	runner.Defer(func() {
-		close(s.keyBroadcastCh)
-	})
 	runner.Go(func() error {
 		return s.watchNewEonPubkey(ctx)
 	})
 	return nil
 }
 
+func (s *EonPubKeySyncer) getInitialPubKeys(ctx context.Context) ([]*event.EonPublicKey, error) {
+	// This blocknumber specifies AT what state
+	// the contract is called
+	// XXX: does the call-opts blocknumber -1 also means latest?
+	opts := &bind.CallOpts{
+		Context:     ctx,
+		BlockNumber: s.StartBlock.Int,
+	}
+	numKS, err := s.ksManager.GetNumKeyperSets(opts)
+	if err != nil {
+		return nil, err
+	}
+	// this blocknumber specifies the argument to the contract
+	// getter
+	activeEon, err := s.ksManager.GetKeyperSetIndexByBlock(opts, s.StartBlock.Uint64())
+	if err != nil {
+		return nil, err
+	}
+
+	initialPubKeys := []*event.EonPublicKey{}
+	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)
+	}
+	return initialPubKeys, nil
+}
+
 func (s *EonPubKeySyncer) logCallError(attrName string, err error) {
 	s.Log.Error(
 		fmt.Sprintf("could not retrieve `%s` from contract", attrName),