feat(sensor): track if block headers are fetched as parents

Author: minhd-vu

cmd/p2p/sensor/sensor.go

@@ -71,6 +71,8 @@ type (
 		NoDiscovery                  bool
 		MaxRequests                  int
 		RequestsCacheTTL             time.Duration
+		MaxParents                   int
+		ParentsCacheTTL              time.Duration
 		MaxBlocks                    int
 		BlocksCacheTTL               time.Duration
 
@@ -214,6 +216,8 @@ var SensorCmd = &cobra.Command{
 			MsgCounter:       msgCounter,
 			MaxRequests:      inputSensorParams.MaxRequests,
 			RequestsCacheTTL: inputSensorParams.RequestsCacheTTL,
+			MaxParents:       inputSensorParams.MaxParents,
+			ParentsCacheTTL:  inputSensorParams.ParentsCacheTTL,
 		}
 
 		config := ethp2p.Config{
@@ -488,6 +492,8 @@ 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.MaxParents, "max-parents", 1024, "maximum parent block hashes to track per peer (0 for no limit)")
+	f.DurationVar(&inputSensorParams.ParentsCacheTTL, "parents-cache-ttl", 5*time.Minute, "time to live for parent hash 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)")
 }

p2p/cache.go

@@ -187,15 +187,20 @@ func (c *Cache[K, V]) Contains(key K) bool {
 	return true
 }
 
-// Remove removes a key from the cache.
-func (c *Cache[K, V]) Remove(key K) {
+// Remove removes a key from the cache and returns the value if it existed.
+func (c *Cache[K, V]) Remove(key K) (V, bool) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
 	if elem, ok := c.items[key]; ok {
+		e := elem.Value.(*entry[K, V])
 		c.list.Remove(elem)
 		delete(c.items, key)
+		return e.value, true
 	}
+
+	var zero V
+	return zero, false
 }
 
 // Len returns the number of items in the cache.

p2p/database/database.go

@@ -21,8 +21,9 @@ type Database interface {
 	WriteBlock(context.Context, *enode.Node, *types.Block, *big.Int, time.Time)
 
 	// WriteBlockHeaders will write the block headers if ShouldWriteBlocks
-	// returns true.
-	WriteBlockHeaders(context.Context, []*types.Header, time.Time)
+	// returns true. The isParent parameter indicates if these headers were
+	// fetched as parent blocks.
+	WriteBlockHeaders(context.Context, []*types.Header, time.Time, bool)
 
 	// WriteBlockHashes will write the block hashes if ShouldWriteBlockEvents
 	// returns true.

p2p/database/datastore.go

@@ -75,6 +75,7 @@ type DatastoreHeader struct {
 	BaseFee       string
 	TimeFirstSeen time.Time
 	TTL           time.Time
+	IsParent      bool
 }
 
 // DatastoreBlock represents a block stored in datastore.
@@ -184,15 +185,16 @@ func (d *Datastore) WriteBlock(ctx context.Context, peer *enode.Node, block *typ
 // WriteBlockHeaders will write the block headers to datastore. It will not
 // write block events because headers will only be sent to the sensor when
 // requested. The block events will be written when the hash is received
-// instead.
-func (d *Datastore) WriteBlockHeaders(ctx context.Context, headers []*types.Header, tfs time.Time) {
+// instead. The isParent parameter indicates if these headers were fetched
+// as parent blocks.
+func (d *Datastore) WriteBlockHeaders(ctx context.Context, headers []*types.Header, tfs time.Time, isParent bool) {
 	if d.client == nil || !d.ShouldWriteBlocks() {
 		return
 	}
 
 	for _, h := range headers {
 		d.runAsync(func() {
-			d.writeBlockHeader(ctx, h, tfs)
+			d.writeBlockHeader(ctx, h, tfs, isParent)
 		})
 	}
 }
@@ -314,7 +316,7 @@ func (d *Datastore) HasBlock(ctx context.Context, hash common.Hash) bool {
 
 // newDatastoreHeader creates a DatastoreHeader from a types.Header. Some
 // values are converted into strings to prevent a loss of precision.
-func (d *Datastore) newDatastoreHeader(header *types.Header, tfs time.Time) *DatastoreHeader {
+func (d *Datastore) newDatastoreHeader(header *types.Header, tfs time.Time, isParent bool) *DatastoreHeader {
 	return &DatastoreHeader{
 		ParentHash:    datastore.NameKey(BlocksKind, header.ParentHash.Hex(), nil),
 		UncleHash:     header.UncleHash.Hex(),
@@ -334,6 +336,7 @@ func (d *Datastore) newDatastoreHeader(header *types.Header, tfs time.Time) *Dat
 		BaseFee:       header.BaseFee.String(),
 		TimeFirstSeen: tfs,
 		TTL:           tfs.Add(d.ttl),
+		IsParent:      isParent,
 	}
 }
 
@@ -390,7 +393,7 @@ func (d *Datastore) writeBlock(ctx context.Context, block *types.Block, td *big.
 
 		if dsBlock.DatastoreHeader == nil {
 			shouldWrite = true
-			dsBlock.DatastoreHeader = d.newDatastoreHeader(block.Header(), tfs)
+			dsBlock.DatastoreHeader = d.newDatastoreHeader(block.Header(), tfs, false)
 		}
 
 		if len(dsBlock.TotalDifficulty) == 0 {
@@ -414,7 +417,7 @@ func (d *Datastore) writeBlock(ctx context.Context, block *types.Block, td *big.
 			shouldWrite = true
 			dsBlock.Uncles = make([]*datastore.Key, 0, len(block.Uncles()))
 			for _, uncle := range block.Uncles() {
-				d.writeBlockHeader(ctx, uncle, tfs)
+				d.writeBlockHeader(ctx, uncle, tfs, false)
 				dsBlock.Uncles = append(dsBlock.Uncles, datastore.NameKey(BlocksKind, uncle.Hash().Hex(), nil))
 			}
 		}
@@ -475,8 +478,8 @@ func (d *Datastore) writeEvents(ctx context.Context, peer *enode.Node, eventKind
 }
 
 // writeBlockHeader will write the block header to datastore if it doesn't
-// exist.
-func (d *Datastore) writeBlockHeader(ctx context.Context, header *types.Header, tfs time.Time) {
+// exist. The isParent parameter indicates if this block was fetched as a parent block.
+func (d *Datastore) writeBlockHeader(ctx context.Context, header *types.Header, tfs time.Time, isParent bool) {
 	key := datastore.NameKey(BlocksKind, header.Hash().Hex(), nil)
 
 	_, err := d.client.RunInTransaction(ctx, func(tx *datastore.Transaction) error {
@@ -485,7 +488,7 @@ func (d *Datastore) writeBlockHeader(ctx context.Context, header *types.Header,
 			return nil
 		}
 
-		block.DatastoreHeader = d.newDatastoreHeader(header, tfs)
+		block.DatastoreHeader = d.newDatastoreHeader(header, tfs, isParent)
 		_, err := tx.Put(key, &block)
 		return err
 	}, datastore.MaxAttempts(MaxAttempts))
@@ -522,7 +525,7 @@ func (d *Datastore) writeBlockBody(ctx context.Context, body *eth.BlockBody, has
 			shouldWrite = true
 			block.Uncles = make([]*datastore.Key, 0, len(body.Uncles))
 			for _, uncle := range body.Uncles {
-				d.writeBlockHeader(ctx, uncle, tfs)
+				d.writeBlockHeader(ctx, uncle, tfs, false)
 				block.Uncles = append(block.Uncles, datastore.NameKey(BlocksKind, uncle.Hash().Hex(), nil))
 			}
 		}

p2p/database/json.go

@@ -69,6 +69,7 @@ type JSONBlock struct {
 	TxCount         int       `json:"tx_count"`
 	UncleCount      int       `json:"uncle_count"`
 	TimeFirstSeen   time.Time `json:"time_first_seen"`
+	IsParent        bool      `json:"is_parent"`
 }
 
 // JSONBlockEvent represents a block event in JSON format.
@@ -179,7 +180,8 @@ func (j *JSONDatabase) writeBlock(block *types.Block, td *big.Int, tfs time.Time
 }
 
 // WriteBlockHeaders writes the block headers as JSON.
-func (j *JSONDatabase) WriteBlockHeaders(ctx context.Context, headers []*types.Header, tfs time.Time) {
+// The isParent parameter indicates if these headers were fetched as parent blocks.
+func (j *JSONDatabase) WriteBlockHeaders(ctx context.Context, headers []*types.Header, tfs time.Time, isParent bool) {
 	if !j.ShouldWriteBlocks() {
 		return
 	}
@@ -196,6 +198,7 @@ func (j *JSONDatabase) WriteBlockHeaders(ctx context.Context, headers []*types.H
 			GasUsed:       header.GasUsed,
 			Difficulty:    header.Difficulty.String(),
 			TimeFirstSeen: tfs,
+			IsParent:      isParent,
 		}
 
 		if header.BaseFee != nil {

p2p/database/nodb.go

@@ -26,7 +26,7 @@ func (n *nodb) WriteBlock(ctx context.Context, peer *enode.Node, block *types.Bl
 }
 
 // WriteBlockHeaders does nothing.
-func (n *nodb) WriteBlockHeaders(ctx context.Context, headers []*types.Header, tfs time.Time) {
+func (n *nodb) WriteBlockHeaders(ctx context.Context, headers []*types.Header, tfs time.Time, isParent bool) {
 }
 
 // WriteBlockHashes does nothing.

p2p/protocol.go

@@ -47,6 +47,10 @@ type conn struct {
 	requests   *Cache[uint64, common.Hash]
 	requestNum uint64
 
+	// parents tracks hashes of blocks requested as parents to mark them
+	// with IsParent=true when writing to the database.
+	parents *Cache[common.Hash, struct{}]
+
 	// conns provides access to the global connection manager, which includes
 	// the blocks cache shared across all peers.
 	conns *Conns
@@ -83,6 +87,10 @@ type EthProtocolOptions struct {
 	// Requests cache configuration
 	MaxRequests      int
 	RequestsCacheTTL time.Duration
+
+	// Parent hash tracking cache configuration
+	MaxParents      int
+	ParentsCacheTTL time.Duration
 }
 
 // HeadBlock contains the necessary head block data for the status message.
@@ -110,6 +118,7 @@ func NewEthProtocol(version uint, opts EthProtocolOptions) ethp2p.Protocol {
 				db:           opts.Database,
 				requests:     NewCache[uint64, common.Hash](opts.MaxRequests, opts.RequestsCacheTTL),
 				requestNum:   0,
+				parents:      NewCache[common.Hash, struct{}](opts.MaxParents, opts.ParentsCacheTTL),
 				head:         opts.Head,
 				headMutex:    opts.HeadMutex,
 				counter:      opts.MsgCounter,
@@ -277,8 +286,9 @@ func (c *conn) readStatus(packet *eth.StatusPacket) error {
 
 // getBlockData will send GetBlockHeaders and/or GetBlockBodies requests to the
 // peer based on what parts of the block we already have. It will return an error
-// if sending either of the requests failed.
-func (c *conn) getBlockData(hash common.Hash, cache BlockCache) error {
+// if sending either of the requests failed. The isParent parameter indicates if
+// this block is being fetched as a parent block.
+func (c *conn) getBlockData(hash common.Hash, cache BlockCache, isParent bool) error {
 	// Only request header if we don't have it
 	if cache.Header == nil {
 		headersRequest := &GetBlockHeaders{
@@ -290,6 +300,10 @@ func (c *conn) getBlockData(hash common.Hash, cache BlockCache) error {
 			},
 		}
 
+		if isParent {
+			c.parents.Add(hash, struct{}{})
+		}
+
 		c.countMsgSent(headersRequest.Name(), 1)
 		if err := ethp2p.Send(c.rw, eth.GetBlockHeadersMsg, headersRequest); err != nil {
 			return err
@@ -343,7 +357,7 @@ func (c *conn) getParentBlock(ctx context.Context, header *types.Header) error {
 		Str("number", new(big.Int).Sub(header.Number, big.NewInt(1)).String()).
 		Msg("Fetching missing parent block")
 
-	return c.getBlockData(header.ParentHash, cache)
+	return c.getBlockData(header.ParentHash, cache, true)
 }
 
 func (c *conn) handleNewBlockHashes(ctx context.Context, msg ethp2p.Msg) error {
@@ -369,7 +383,7 @@ func (c *conn) handleNewBlockHashes(ctx context.Context, msg ethp2p.Msg) error {
 		}
 
 		// Request only the parts we don't have
-		if err := c.getBlockData(hash, cache); err != nil {
+		if err := c.getBlockData(hash, cache, false); err != nil {
 			return err
 		}
 
@@ -422,6 +436,10 @@ func (c *conn) handleBlockHeaders(ctx context.Context, msg ethp2p.Msg) error {
 	tfs := time.Now()
 
 	headers := packet.BlockHeadersRequest
+	if len(headers) == 0 {
+		return nil
+	}
+
 	c.countMsgReceived(packet.Name(), float64(len(headers)))
 
 	for _, header := range headers {
@@ -430,7 +448,10 @@ func (c *conn) handleBlockHeaders(ctx context.Context, msg ethp2p.Msg) error {
 		}
 	}
 
-	c.db.WriteBlockHeaders(ctx, headers, tfs)
+	// Check if any of these headers were requested as parent blocks
+	_, isParent := c.parents.Remove(headers[0].Hash())
+
+	c.db.WriteBlockHeaders(ctx, headers, tfs, isParent)
 
 	// Update cache to store headers
 	for _, header := range headers {