move update limiter to listener

Author: angaz

pkg/database/execution_crawler.go

@@ -399,58 +399,6 @@ func (db *DB) InsertBlocks(
 	return nil
 }
 
-type UpdateLimiter struct {
-	m    map[string]time.Time
-	ttl  time.Duration
-	lock sync.Mutex
-}
-
-func NewUpdateLimiter(ttl time.Duration) *UpdateLimiter {
-	limiter := &UpdateLimiter{
-		m:    map[string]time.Time{},
-		ttl:  ttl,
-		lock: sync.Mutex{},
-	}
-
-	go limiter.runCleaner()
-
-	return limiter
-}
-
-func (l *UpdateLimiter) runCleaner() {
-	for {
-		time.Sleep(time.Minute)
-
-		l.lock.Lock()
-
-		for key, ts := range l.m {
-			if time.Since(ts) > l.ttl {
-				delete(l.m, key)
-			}
-		}
-
-		l.lock.Unlock()
-	}
-}
-
-func (l *UpdateLimiter) IsLimited(node common.NodeJSON) bool {
-	l.lock.Lock()
-	defer l.lock.Unlock()
-
-	id := node.IDString()
-
-	t, found := l.m[id]
-	if found && time.Since(t) < l.ttl {
-		return true
-	}
-
-	l.m[id] = time.Now()
-
-	return false
-}
-
-var limiter = NewUpdateLimiter(3 * time.Hour)
-
 func (db *DB) UpsertCrawledNode(ctx context.Context, tx pgx.Tx, node common.NodeJSON) error {
 	defer metrics.NodeUpdateInc(node.Direction.String(), node.Error)
 
@@ -459,10 +407,6 @@ func (db *DB) UpsertCrawledNode(ctx context.Context, tx pgx.Tx, node common.Node
 		return nil
 	}
 
-	if limiter.IsLimited(node) {
-		return nil
-	}
-
 	if !node.EthNode {
 		err := db.UpdateNotEthNode(ctx, tx, node)
 		if err != nil {

pkg/execution/listener/listener.go

@@ -3,13 +3,15 @@ package listener
 import (
 	"context"
 	"crypto/ecdsa"
+	"encoding/hex"
 	"fmt"
 	"net"
 	"sync"
 	"time"
 
 	"log/slog"
 
+	ethp2p "github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/enode"
 	"github.com/ethereum/go-ethereum/p2p/netutil"
 	"github.com/ethereum/node-crawler/pkg/common"
@@ -138,6 +140,58 @@ func (l *Listener) startListener(ctx context.Context, nodeKey *ecdsa.PrivateKey,
 	}()
 }
 
+type UpdateLimiter struct {
+	m    map[string]time.Time
+	ttl  time.Duration
+	lock sync.Mutex
+}
+
+func NewUpdateLimiter(ttl time.Duration) *UpdateLimiter {
+	limiter := &UpdateLimiter{
+		m:    map[string]time.Time{},
+		ttl:  ttl,
+		lock: sync.Mutex{},
+	}
+
+	go limiter.runCleaner()
+
+	return limiter
+}
+
+func (l *UpdateLimiter) runCleaner() {
+	for {
+		time.Sleep(time.Minute)
+
+		l.lock.Lock()
+
+		for key, ts := range l.m {
+			if time.Since(ts) > l.ttl {
+				delete(l.m, key)
+			}
+		}
+
+		l.lock.Unlock()
+	}
+}
+
+func (l *UpdateLimiter) IsLimited(pubkey *ecdsa.PublicKey) bool {
+	l.lock.Lock()
+	defer l.lock.Unlock()
+
+	id := hex.EncodeToString(common.PubkeyBytes(pubkey))
+
+	t, found := l.m[id]
+	if found && time.Since(t) < l.ttl {
+		return true
+	}
+
+	l.m[id] = time.Now()
+
+	return false
+}
+
+var limiter = NewUpdateLimiter(3 * time.Hour)
+
 func (l *Listener) crawlPeer(ctx context.Context, nodeKey *ecdsa.PrivateKey, fd net.Conn) {
 	pubKey, conn, err := p2p.Accept(nodeKey, fd)
 	if err != nil {
@@ -150,6 +204,12 @@ func (l *Listener) crawlPeer(ctx context.Context, nodeKey *ecdsa.PrivateKey, fd
 	}
 	defer conn.Close()
 
+	if limiter.IsLimited(pubKey) {
+		_ = conn.Write(p2p.Disconnect{Reason: ethp2p.DiscTooManyPeers})
+
+		return
+	}
+
 	err = l.db.WithTxAsync(
 		ctx,
 		database.TxOptionsDeferrable,