feat(sensor): use cache to track seen blocks (#747)

* feat(sensor): use LRU with TTL for requests cache

* docs: make gen-doc

* fix: rename requests cache variables

* fix: comment

* fix: requests cache opts name

* feat(sensor): use cache to track seen blocks

* docs: make gen-doc

* feat: global blocks cache

* fix: rename method

* fix: conditional sends?

* fix: flags

* chore: use conns opts

* fix: revert request spelling

* fix: remove wrong check

* fix: logic issues

* feat: store entire blocks

* chore: runAsync refactor

* feat: optimizations

Author: minhd-vu

cmd/p2p/sensor/sensor.go

@@ -71,6 +71,8 @@ type (
 		NoDiscovery                  bool
 		MaxRequests                  int
 		RequestsCacheTTL             time.Duration
+		MaxBlocks                    int
+		BlocksCacheTTL               time.Duration
 
 		bootnodes    []*enode.Node
 		staticNodes  []*enode.Node
@@ -192,7 +194,11 @@ var SensorCmd = &cobra.Command{
 		}, []string{"message", "url", "name", "direction"})
 
 		// Create peer connection manager for broadcasting transactions
-		conns := p2p.NewConns()
+		// and managing the global blocks cache
+		conns := p2p.NewConns(p2p.ConnsOptions{
+			MaxBlocks:      inputSensorParams.MaxBlocks,
+			BlocksCacheTTL: inputSensorParams.BlocksCacheTTL,
+		})
 
 		opts := p2p.EthProtocolOptions{
 			Context:          cmd.Context(),
@@ -482,4 +488,6 @@ will result in less chance of missing data but can significantly increase memory
 	f.BoolVar(&inputSensorParams.NoDiscovery, "no-discovery", false, "disable P2P peer discovery")
 	f.IntVar(&inputSensorParams.MaxRequests, "max-requests", 2048, "maximum request IDs to track per peer (0 for no limit)")
 	f.DurationVar(&inputSensorParams.RequestsCacheTTL, "requests-cache-ttl", 5*time.Minute, "time to live for requests cache entries (0 for no expiration)")
+	f.IntVar(&inputSensorParams.MaxBlocks, "max-blocks", 1024, "maximum blocks to track across all peers (0 for no limit)")
+	f.DurationVar(&inputSensorParams.BlocksCacheTTL, "blocks-cache-ttl", 10*time.Minute, "time to live for block cache entries (0 for no expiration)")
 }

doc/polycli_p2p_sensor.md

@@ -24,6 +24,7 @@ If no nodes.json file exists, it will be created.
 
 ```bash
       --api-port uint                 port API server will listen on (default 8080)
+      --blocks-cache-ttl duration     time to live for block cache entries (0 for no expiration) (default 10m0s)
   -b, --bootnodes string              comma separated nodes used for bootstrapping
       --database string               which database to persist data to, options are:
                                         - datastore (GCP Datastore)
@@ -38,6 +39,7 @@ If no nodes.json file exists, it will be created.
   -h, --help                          help for sensor
       --key string                    hex-encoded private key (cannot be set with --key-file)
   -k, --key-file string               private key file (cannot be set with --key)
+      --max-blocks int                maximum blocks to track across all peers (0 for no limit) (default 1024)
   -D, --max-db-concurrency int        maximum number of concurrent database operations to perform (increasing this
                                       will result in less chance of missing data but can significantly increase memory usage) (default 10000)
   -m, --max-peers int                 maximum number of peers to connect to (default 2000)

p2p/cache.go

@@ -94,6 +94,79 @@ func (c *Cache[K, V]) Get(key K) (V, bool) {
 	return e.value, true
 }
 
+// Peek retrieves a value from the cache without updating LRU ordering.
+// Uses a read lock for better concurrency.
+func (c *Cache[K, V]) Peek(key K) (V, bool) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	elem, ok := c.items[key]
+	if !ok {
+		var zero V
+		return zero, false
+	}
+
+	e := elem.Value.(*entry[K, V])
+
+	if c.ttl > 0 && time.Now().After(e.expiresAt) {
+		var zero V
+		return zero, false
+	}
+
+	return e.value, true
+}
+
+// Update atomically updates a value in the cache using the provided update function.
+// The update function receives the current value (or zero value if not found) and
+// returns the new value to store. This is thread-safe and prevents race conditions
+// in get-modify-add patterns.
+func (c *Cache[K, V]) Update(key K, updateFn func(V) V) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	now := time.Now()
+	expiresAt := time.Time{}
+	if c.ttl > 0 {
+		expiresAt = now.Add(c.ttl)
+	}
+
+	var currentVal V
+	if elem, ok := c.items[key]; ok {
+		e := elem.Value.(*entry[K, V])
+		if c.ttl == 0 || !now.After(e.expiresAt) {
+			currentVal = e.value
+			// Update existing entry
+			c.list.MoveToFront(elem)
+			e.value = updateFn(currentVal)
+			e.expiresAt = expiresAt
+			return
+		}
+		// Entry expired, remove it
+		c.list.Remove(elem)
+		delete(c.items, key)
+	}
+
+	// Create new entry
+	newVal := updateFn(currentVal)
+	e := &entry[K, V]{
+		key:       key,
+		value:     newVal,
+		expiresAt: expiresAt,
+	}
+	elem := c.list.PushFront(e)
+	c.items[key] = elem
+
+	// Enforce size limit
+	if c.maxSize > 0 && c.list.Len() > c.maxSize {
+		back := c.list.Back()
+		if back != nil {
+			c.list.Remove(back)
+			e := back.Value.(*entry[K, V])
+			delete(c.items, e.key)
+		}
+	}
+}
+
 // Contains checks if a key exists in the cache and is not expired.
 // Uses a read lock and doesn't update LRU ordering.
 func (c *Cache[K, V]) Contains(key K) bool {

p2p/conns.go

@@ -4,22 +4,37 @@ import (
 	"sync"
 	"time"
 
+	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/eth/protocols/eth"
 	ethp2p "github.com/ethereum/go-ethereum/p2p"
 	"github.com/ethereum/go-ethereum/p2p/enode"
 )
 
 // Conns manages a collection of active peer connections for transaction broadcasting.
+// It also maintains a global cache of blocks written to the database.
 type Conns struct {
 	conns map[string]*conn
 	mu    sync.RWMutex
+
+	// blocks tracks blocks written to the database across all peers
+	// to avoid duplicate writes and requests.
+	blocks *Cache[common.Hash, BlockCache]
+}
+
+// ConnsOptions contains configuration options for creating a new Conns manager.
+type ConnsOptions struct {
+	// MaxBlocks is the maximum number of blocks to track in the cache.
+	MaxBlocks int
+	// BlocksCacheTTL is the time to live for block cache entries.
+	BlocksCacheTTL time.Duration
 }
 
-// NewConns creates a new connection manager.
-func NewConns() *Conns {
+// NewConns creates a new connection manager with a blocks cache.
+func NewConns(opts ConnsOptions) *Conns {
 	return &Conns{
-		conns: make(map[string]*conn),
+		conns:  make(map[string]*conn),
+		blocks: NewCache[common.Hash, BlockCache](opts.MaxBlocks, opts.BlocksCacheTTL),
 	}
 }
 
@@ -91,3 +106,8 @@ func (c *Conns) GetPeerConnectedAt(peerID string) time.Time {
 
 	return time.Time{}
 }
+
+// Blocks returns the global blocks cache.
+func (c *Conns) Blocks() *Cache[common.Hash, BlockCache] {
+	return c.blocks
+}

p2p/database/datastore.go

@@ -152,26 +152,32 @@ func NewDatastore(ctx context.Context, opts DatastoreOptions) Database {
 	}
 }
 
+// runAsync executes the provided function asynchronously with concurrency control.
+// It uses the jobs channel as a semaphore to limit concurrent operations.
+func (d *Datastore) runAsync(fn func()) {
+	d.jobs <- struct{}{}
+	go func() {
+		fn()
+		<-d.jobs
+	}()
+}
+
 // WriteBlock writes the block and the block event to datastore.
 func (d *Datastore) WriteBlock(ctx context.Context, peer *enode.Node, block *types.Block, td *big.Int, tfs time.Time) {
 	if d.client == nil {
 		return
 	}
 
 	if d.ShouldWriteBlockEvents() {
-		d.jobs <- struct{}{}
-		go func() {
+		d.runAsync(func() {
 			d.writeEvent(peer, BlockEventsKind, block.Hash(), BlocksKind, tfs)
-			<-d.jobs
-		}()
+		})
 	}
 
 	if d.ShouldWriteBlocks() {
-		d.jobs <- struct{}{}
-		go func() {
+		d.runAsync(func() {
 			d.writeBlock(ctx, block, td, tfs)
-			<-d.jobs
-		}()
+		})
 	}
 }
 
@@ -185,11 +191,9 @@ func (d *Datastore) WriteBlockHeaders(ctx context.Context, headers []*types.Head
 	}
 
 	for _, h := range headers {
-		d.jobs <- struct{}{}
-		go func(header *types.Header) {
-			d.writeBlockHeader(ctx, header, tfs)
-			<-d.jobs
-		}(h)
+		d.runAsync(func() {
+			d.writeBlockHeader(ctx, h, tfs)
+		})
 	}
 }
 
@@ -203,11 +207,9 @@ func (d *Datastore) WriteBlockBody(ctx context.Context, body *eth.BlockBody, has
 		return
 	}
 
-	d.jobs <- struct{}{}
-	go func() {
+	d.runAsync(func() {
 		d.writeBlockBody(ctx, body, hash, tfs)
-		<-d.jobs
-	}()
+	})
 }
 
 // WriteBlockHashes will write the block events to datastore.
@@ -216,11 +218,9 @@ func (d *Datastore) WriteBlockHashes(ctx context.Context, peer *enode.Node, hash
 		return
 	}
 
-	d.jobs <- struct{}{}
-	go func() {
+	d.runAsync(func() {
 		d.writeEvents(ctx, peer, BlockEventsKind, hashes, BlocksKind, tfs)
-		<-d.jobs
-	}()
+	})
 }
 
 // WriteTransactions will write the transactions and transaction events to datastore.
@@ -230,11 +230,9 @@ func (d *Datastore) WriteTransactions(ctx context.Context, peer *enode.Node, txs
 	}
 
 	if d.ShouldWriteTransactions() {
-		d.jobs <- struct{}{}
-		go func() {
+		d.runAsync(func() {
 			d.writeTransactions(ctx, txs, tfs)
-			<-d.jobs
-		}()
+		})
 	}
 
 	if d.ShouldWriteTransactionEvents() {
@@ -243,11 +241,9 @@ func (d *Datastore) WriteTransactions(ctx context.Context, peer *enode.Node, txs
 			hashes = append(hashes, tx.Hash())
 		}
 
-		d.jobs <- struct{}{}
-		go func() {
+		d.runAsync(func() {
 			d.writeEvents(ctx, peer, TransactionEventsKind, hashes, TransactionsKind, tfs)
-			<-d.jobs
-		}()
+		})
 	}
 }
 
@@ -257,9 +253,7 @@ func (d *Datastore) WritePeers(ctx context.Context, peers []*p2p.Peer, tls time.
 		return
 	}
 
-	d.jobs <- struct{}{}
-	go func() {
-
+	d.runAsync(func() {
 		keys := make([]*datastore.Key, 0, len(peers))
 		dsPeers := make([]*DatastorePeer, 0, len(peers))
 
@@ -279,9 +273,7 @@ func (d *Datastore) WritePeers(ctx context.Context, peers []*p2p.Peer, tls time.
 		if err != nil {
 			log.Error().Err(err).Msg("Failed to write peers")
 		}
-
-		<-d.jobs
-	}()
+	})
 }
 
 func (d *Datastore) MaxConcurrentWrites() int {