Show latency

Author: jannikluhn

rolling-shutter/gnosiskeyperwatcher/blocks.go

@@ -2,6 +2,7 @@ package gnosiskeyperwatcher
 
 import (
 	"context"
+	"time"
 
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethclient"
@@ -12,12 +13,19 @@ import (
 )
 
 type BlocksWatcher struct {
-	config *keyper.Config
+	config        *keyper.Config
+	blocksChannel chan *BlockReceivedEvent
 }
 
-func NewBlocksWatcher(config *keyper.Config) *BlocksWatcher {
+type BlockReceivedEvent struct {
+	Header *types.Header
+	Time   time.Time
+}
+
+func NewBlocksWatcher(config *keyper.Config, blocksChannel chan *BlockReceivedEvent) *BlocksWatcher {
 	return &BlocksWatcher{
-		config: config,
+		config:        config,
+		blocksChannel: blocksChannel,
 	}
 }
 
@@ -41,6 +49,11 @@ func (w *BlocksWatcher) Start(ctx context.Context, runner service.Runner) error
 				return ctx.Err()
 			case head := <-newHeads:
 				w.logNewHead(head)
+				ev := &BlockReceivedEvent{
+					Header: head,
+					Time:   time.Now(),
+				}
+				w.blocksChannel <- ev
 			case err := <-sub.Err():
 				return err
 			}

rolling-shutter/gnosiskeyperwatcher/keys.go

@@ -2,6 +2,9 @@ package gnosiskeyperwatcher
 
 import (
 	"context"
+	"fmt"
+	"sync"
+	"time"
 
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/rs/zerolog/log"
@@ -13,22 +16,33 @@ import (
 )
 
 type KeysWatcher struct {
-	config *keyper.Config
+	config        *keyper.Config
+	blocksChannel chan *BlockReceivedEvent
+
+	recentBlocksMux sync.Mutex
+	recentBlocks    map[uint64]*BlockReceivedEvent
+	mostRecentBlock uint64
 }
 
-func NewKeysWatcher(config *keyper.Config) *KeysWatcher {
+func NewKeysWatcher(config *keyper.Config, blocksChannel chan *BlockReceivedEvent) *KeysWatcher {
 	return &KeysWatcher{
-		config: config,
+		config:          config,
+		blocksChannel:   blocksChannel,
+		recentBlocksMux: sync.Mutex{},
+		recentBlocks:    make(map[uint64]*BlockReceivedEvent),
+		mostRecentBlock: 0,
 	}
 }
 
-func (w *KeysWatcher) Start(_ context.Context, runner service.Runner) error {
+func (w *KeysWatcher) Start(ctx context.Context, runner service.Runner) error {
 	p2pService, err := p2p.New(w.config.P2P)
 	if err != nil {
 		return err
 	}
 	p2pService.AddMessageHandler(w)
 
+	runner.Go(func() error { return w.insertBlocks(ctx) })
+
 	return runner.StartService(p2pService)
 }
 
@@ -43,11 +57,70 @@ func (w *KeysWatcher) ValidateMessage(_ context.Context, _ p2pmsg.Message) (pubs
 }
 
 func (w *KeysWatcher) HandleMessage(_ context.Context, msgUntyped p2pmsg.Message) ([]p2pmsg.Message, error) {
+	t := time.Now()
 	msg := msgUntyped.(*p2pmsg.DecryptionKeys)
 	extra := msg.Extra.(*p2pmsg.DecryptionKeys_Gnosis).Gnosis
+
+	ev, ok := w.getRecentBlock(extra.Slot)
+	if !ok {
+		log.Warn().
+			Uint64("keys-block", extra.Slot).
+			Uint64("most-recent-block", w.mostRecentBlock).
+			Msg("received keys for unknown block")
+		return []p2pmsg.Message{}, nil
+	}
+
+	dt := t.Sub(ev.Time)
 	log.Info().
 		Uint64("block", extra.Slot).
 		Int("num-keys", len(msg.Keys)).
+		Str("latency", fmt.Sprintf("%.2fs", dt.Seconds())).
 		Msg("new keys")
 	return []p2pmsg.Message{}, nil
 }
+
+func (w *KeysWatcher) insertBlocks(ctx context.Context) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case ev, ok := <-w.blocksChannel:
+			if !ok {
+				return nil
+			}
+			w.insertBlock(ev)
+			w.clearOldBlocks(ev)
+		}
+	}
+}
+
+func (w *KeysWatcher) insertBlock(ev *BlockReceivedEvent) {
+	w.recentBlocksMux.Lock()
+	defer w.recentBlocksMux.Unlock()
+	w.recentBlocks[ev.Header.Number.Uint64()] = ev
+	if ev.Header.Number.Uint64() > w.mostRecentBlock {
+		w.mostRecentBlock = ev.Header.Number.Uint64()
+	}
+}
+
+func (w *KeysWatcher) clearOldBlocks(latestEv *BlockReceivedEvent) {
+	w.recentBlocksMux.Lock()
+	defer w.recentBlocksMux.Unlock()
+
+	tooOld := []uint64{}
+	for block := range w.recentBlocks {
+		if block < latestEv.Header.Number.Uint64()-100 {
+			tooOld = append(tooOld, block)
+		}
+	}
+	for _, block := range tooOld {
+		delete(w.recentBlocks, block)
+	}
+}
+
+func (w *KeysWatcher) getRecentBlock(blockNumber uint64) (*BlockReceivedEvent, bool) {
+	w.recentBlocksMux.Lock()
+	defer w.recentBlocksMux.Unlock()
+	ev, ok := w.recentBlocks[blockNumber]
+	return ev, ok
+}

rolling-shutter/gnosiskeyperwatcher/watcher.go

@@ -18,7 +18,8 @@ func New(config *keyper.Config) *Watcher {
 }
 
 func (w *Watcher) Start(_ context.Context, runner service.Runner) error {
-	blocksWatcher := NewBlocksWatcher(w.config)
-	keysWatcher := NewKeysWatcher(w.config)
+	blocksChannel := make(chan *BlockReceivedEvent)
+	blocksWatcher := NewBlocksWatcher(w.config, blocksChannel)
+	keysWatcher := NewKeysWatcher(w.config, blocksChannel)
 	return runner.StartService(blocksWatcher, keysWatcher)
 }