From 32603b7ea40b97b13110f996687f66925f5e57e8 Mon Sep 17 00:00:00 2001 From: healthykim Date: Thu, 9 Jul 2026 18:13:43 +0200 Subject: [PATCH] apply review from claude --- core/txpool/blobpool/buffer.go | 1 + eth/fetcher/blob_fetcher.go | 71 ++++++++++++++++++++++++++-------- eth/protocols/eth/broadcast.go | 3 ++ eth/protocols/eth/protocol.go | 2 +- 4 files changed, 60 insertions(+), 17 deletions(-) diff --git a/core/txpool/blobpool/buffer.go b/core/txpool/blobpool/buffer.go index 18a5811ee7..d80e793b82 100644 --- a/core/txpool/blobpool/buffer.go +++ b/core/txpool/blobpool/buffer.go @@ -177,6 +177,7 @@ func (b *BlobBuffer) storeCompleted(hash common.Hash, tx *types.Transaction, cel b.dropPeers(badPeers) delete(b.cells, hash) delete(b.txs, hash) + return } blobCount := len(tx.BlobHashes()) sorted, custody := sortCells(cells, blobCount) diff --git a/eth/fetcher/blob_fetcher.go b/eth/fetcher/blob_fetcher.go index fa3d84b820..9e27581bbc 100644 --- a/eth/fetcher/blob_fetcher.go +++ b/eth/fetcher/blob_fetcher.go @@ -17,6 +17,7 @@ package fetcher import ( + "fmt" "iter" "math/rand" "slices" @@ -93,7 +94,6 @@ type fetchStatus struct { fetching types.CustodyBitmap // To avoid fetching cells which had already been fetched / currently being fetched fetched []uint64 // Custody indices that have been fetched (per-blob, same for all blobs) deliveries map[string]*PeerCellDelivery // Per-peer cell deliveries - blobCount int // Number of blobs in this tx (set on first delivery) } type BlobFetcherFunctions struct { @@ -214,6 +214,19 @@ func (f *BlobFetcher) Notify(peer string, txs []common.Hash, cells types.Custody func (f *BlobFetcher) Enqueue(peer string, hashes []common.Hash, cells [][]kzg4844.Cell, cellBitmap types.CustodyBitmap) error { blobReplyInMeter.Mark(int64(len(hashes))) + if len(hashes) != len(cells) { + return fmt.Errorf("cells reply length mismatch: %d hashes, %d cells", len(hashes), len(cells)) + } + // An empty custody mask means "no cells"; a peer delivering cells alongside it + // is malformed and should be dropped. + if cellBitmap.OneCount() == 0 { + for _, c := range cells { + if len(c) != 0 { + return fmt.Errorf("cells delivered with empty custody mask (%d cells)", len(c)) + } + } + } + select { case f.cleanup <- &payloadDelivery{origin: peer, txs: hashes, cells: cells, cellBitmap: cellBitmap}: case <-f.quit: @@ -450,10 +463,12 @@ func (f *BlobFetcher) loop() { // Allow other candidates to be requested these cells f.fetches[hash].fetching = f.fetches[hash].fetching.Difference(req.cells) - // Drop cells if there is no alternate source to fetch cells from - if len(f.alternates[hash]) == 0 { + // Drop cells if the remaining sources can no longer complete the fetch + if !f.recoverable(hash) { delete(f.alternates, hash) delete(f.fetches, hash) + delete(f.full, hash) + delete(f.partial, hash) } } if len(f.announces[peer]) == 0 { @@ -502,14 +517,8 @@ func (f *BlobFetcher) loop() { continue } indices := delivery.cellBitmap.Indices() - cellsPerBlob := len(indices) - if cellsPerBlob > 0 { + if len(indices) > 0 { status := f.fetches[hash] - blobCount := len(delivery.cells[i]) / cellsPerBlob - if status.blobCount == 0 { - status.blobCount = blobCount - status.deliveries = make(map[string]*PeerCellDelivery) - } status.deliveries[delivery.origin] = &PeerCellDelivery{ Cells: delivery.cells[i], Indices: indices, @@ -555,6 +564,8 @@ func (f *BlobFetcher) loop() { } delete(f.alternates, hash) delete(f.fetches, hash) + delete(f.full, hash) + delete(f.partial, hash) } } blobRequestDoneMeter.Mark(int64(len(delivery.txs))) @@ -635,9 +646,11 @@ func (f *BlobFetcher) loop() { // Undelivered hash, reschedule if there's an alternative origin available f.fetches[hash].fetching = f.fetches[hash].fetching.Difference(req.cells) delete(f.alternates[hash], drop.peer) - if len(f.alternates[hash]) == 0 { + if !f.recoverable(hash) { delete(f.alternates, hash) delete(f.fetches, hash) + delete(f.full, hash) + delete(f.partial, hash) } } } @@ -678,12 +691,12 @@ func (f *BlobFetcher) rescheduleWait(timer *mclock.Timer, trigger chan struct{}) for _, instance := range f.waittime { if earliest > instance { earliest = instance - if txArriveTimeout-time.Duration(now-earliest) < txGatherSlack { + if blobAvailabilityTimeout-time.Duration(now-earliest) < txGatherSlack { break } } } - *timer = f.clock.AfterFunc(txArriveTimeout-time.Duration(now-earliest), func() { + *timer = f.clock.AfterFunc(blobAvailabilityTimeout-time.Duration(now-earliest), func() { trigger <- struct{}{} }) } @@ -742,6 +755,27 @@ func (f *BlobFetcher) consumeToken(peer string, n int) bool { return true } +// recoverable checks whether the missing cells of a tx can be fully +// received from the alternative peers. +func (f *BlobFetcher) recoverable(hash common.Hash) bool { + status := f.fetches[hash] + if status == nil { + // this should not happen + return false + } + // get union of already fetched cells and the cells + // that can be fetched from alternative peers + covered := types.NewCustodyBitmap(status.fetched) + for _, cells := range f.alternates[hash] { + covered = covered.Union(cells) + } + if _, ok := f.full[hash]; ok { + return covered.OneCount() >= kzg4844.DataPerBlob + } + // return false if our custody cannot be covered + return f.custody.Difference(covered).OneCount() == 0 +} + func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{}, whitelist map[string]struct{}) { // Gather the set of peers we want to retrieve from (default to all) actives := whitelist @@ -781,8 +815,9 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{} if unfetched.OneCount() > 0 && f.consumeToken(peer, unfetched.OneCount()) { if f.fetches[hash] == nil { f.fetches[hash] = &fetchStatus{ - fetching: unfetched, - fetched: make([]uint64, 0), + fetching: unfetched, + fetched: make([]uint64, 0), + deliveries: make(map[string]*PeerCellDelivery), } } else { f.fetches[hash].fetching = f.fetches[hash].fetching.Union(unfetched) @@ -829,8 +864,12 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{} request = append(request, cr) } f.requests[peer] = request + // Iterate over a copy: the loop may swap-truncate f.requests[peer] + // concurrently when a delivery for this peer arrives. + reqs := make([]*cellRequest, len(request)) + copy(reqs, request) go func() { - for _, req := range request { + for _, req := range reqs { blobRequestOutMeter.Mark(int64(len(req.txs))) if err := f.fn.FetchPayloads(peer, req.txs, req.cells); err != nil { blobRequestFailMeter.Mark(int64(len(req.txs))) diff --git a/eth/protocols/eth/broadcast.go b/eth/protocols/eth/broadcast.go index 1a934b067b..eb51b1aab9 100644 --- a/eth/protocols/eth/broadcast.go +++ b/eth/protocols/eth/broadcast.go @@ -144,6 +144,9 @@ func (p *Peer) announceTransactions() { } size += common.HashLength + processed[count] = true + } else { + // Transaction is no longer in the pool; drop it from the queue. processed[count] = true } } diff --git a/eth/protocols/eth/protocol.go b/eth/protocols/eth/protocol.go index 46434dc800..2e27fdd74c 100644 --- a/eth/protocols/eth/protocol.go +++ b/eth/protocols/eth/protocol.go @@ -252,7 +252,7 @@ type ReceiptsPacket70 struct { // ReceiptsRLPResponse is used for receipts, when we already have it encoded type ReceiptsRLPResponse []rlp.RawValue -// NewPooledTransactionHashesPacket71 represents a transaction announcement packet on eth/69. +// NewPooledTransactionHashesPacket71 represents a transaction announcement packet on eth/72. type NewPooledTransactionHashesPacket71 struct { Types []byte Sizes []uint32