eth/fetcher: clean up tx hashes that are not retrivable

eth/fetcher/blob_fetcher.go

@@ -327,6 +327,7 @@ func (f *BlobFetcher) loop() {
 				}
 				if _, ok := f.full[hash]; ok {
 					// 1) Decided to send full request of the tx
+					// if the cells are not set to all bitmap, ignore.
 					if ann.cells != types.CustodyBitmapAll {
 						continue
 					}
@@ -343,6 +344,7 @@ func (f *BlobFetcher) loop() {
 				if _, ok := f.partial[hash]; ok {
 					// 2) Decided to send partial request of the tx
 					if f.waitlist[hash] != nil {
+						// it is already waiting for availability check
 						if ann.cells != types.CustodyBitmapAll {
 							// Availability check is only meaningful with full availability announcements
 							continue
@@ -384,7 +386,6 @@ func (f *BlobFetcher) loop() {
 						continue
 					}
 					// Add this peer as a possible fetch source
-					// todo: Did we remove fetch from partial
 					if f.announces[ann.origin] == nil {
 						f.announces[ann.origin] = make(map[common.Hash]*cellWithSeq)
 					}
@@ -460,15 +461,16 @@ func (f *BlobFetcher) loop() {
 							// Do not request the same tx from this peer
 							delete(f.announces[peer], hash)
 							delete(f.alternates[hash], peer)
+							if f.fetches[hash] == nil {
+								// already completed or discarded via another peer
+								continue
+							}
 							// Allow other candidates to be requested these cells
 							f.fetches[hash].fetching = f.fetches[hash].fetching.Difference(req.cells)
 
-							// Drop cells if the remaining sources can no longer complete the fetch
+							// Drop the tx entirely if the remaining sources can no longer complete it
 							if !f.recoverable(hash) {
-								delete(f.alternates, hash)
+								f.discard(hash)
-								delete(f.fetches, hash)
-								delete(f.full, hash)
-								delete(f.partial, hash)
 							}
 						}
 						if len(f.announces[peer]) == 0 {
@@ -512,8 +514,8 @@ func (f *BlobFetcher) loop() {
 			}
 
 			for i, hash := range delivery.txs {
-				if !slices.Contains(request.txs, hash) {
+				if !slices.Contains(request.txs, hash) || f.fetches[hash] == nil {
-					// Unexpected hash, ignore
+					// Unexpected or canceled hash, ignore
 					continue
 				}
 				indices := delivery.cellBitmap.Indices()
@@ -556,16 +558,7 @@ func (f *BlobFetcher) loop() {
 					collectedCustody := types.NewCustodyBitmap(status.fetched)
 					f.fn.AddCells(hash, status.deliveries, collectedCustody)
 
-					for peer, txset := range f.announces {
+					f.discard(hash)
-						delete(txset, hash)
-						if len(txset) == 0 {
-							delete(f.announces, peer)
-						}
-					}
-					delete(f.alternates, hash)
-					delete(f.fetches, hash)
-					delete(f.full, hash)
-					delete(f.partial, hash)
 				}
 			}
 			blobRequestDoneMeter.Mark(int64(len(delivery.txs)))
@@ -643,14 +636,15 @@ func (f *BlobFetcher) loop() {
 			if request, ok := f.requests[drop.peer]; ok && len(request) != 0 {
 				for _, req := range request {
 					for _, hash := range req.txs {
+						if f.fetches[hash] == nil {
+							// already canceled
+							continue
+						}
 						// 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 !f.recoverable(hash) {
-							delete(f.alternates, hash)
+							f.discard(hash)
-							delete(f.fetches, hash)
-							delete(f.full, hash)
-							delete(f.partial, hash)
 						}
 					}
 				}
@@ -758,16 +752,15 @@ func (f *BlobFetcher) consumeToken(peer string, n int) bool {
 // 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]
+	// union of already fetched cells and the cells still announced by peers
-	if status == nil {
+	var covered types.CustodyBitmap
-		// this should not happen
+	if status := f.fetches[hash]; status != nil {
-		return false
+		covered = types.NewCustodyBitmap(status.fetched)
 	}
-	// get union of already fetched cells and the cells
+	for _, anns := range f.announces {
-	// that can be fetched from alternative peers
+		if cs, ok := anns[hash]; ok {
-	covered := types.NewCustodyBitmap(status.fetched)
+			covered = covered.Union(cs.cells)
-	for _, cells := range f.alternates[hash] {
+		}
-		covered = covered.Union(cells)
 	}
 	if _, ok := f.full[hash]; ok {
 		return covered.OneCount() >= kzg4844.DataPerBlob
@@ -776,6 +769,23 @@ func (f *BlobFetcher) recoverable(hash common.Hash) bool {
 	return f.custody.Difference(covered).OneCount() == 0
 }
 
+// discard removes every trace of a transaction from the fetcher: its fetch
+// state and all pending announcements/alternates across peers.
+func (f *BlobFetcher) discard(hash common.Hash) {
+	for peer, anns := range f.announces {
+		if _, ok := anns[hash]; ok {
+			delete(anns, hash)
+			if len(anns) == 0 {
+				delete(f.announces, peer)
+			}
+		}
+	}
+	delete(f.alternates, hash)
+	delete(f.fetches, hash)
+	delete(f.full, hash)
+	delete(f.partial, hash)
+}
+
 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
@@ -801,6 +811,9 @@ func (f *BlobFetcher) scheduleFetches(timer *mclock.Timer, timeout chan struct{}
 			custodies []types.CustodyBitmap
 		)
 		for hash, cells := range f.announcesByArrival(f.announces[peer]) {
+			if !f.recoverable(hash) {
+				continue
+			}
 			var unfetched types.CustodyBitmap
 			if f.fetches[hash] == nil {
 				// tx is not being fetched

eth/fetcher/blob_fetcher_test.go

@@ -639,6 +639,68 @@ func TestBlobFetcherPeerDrop(t *testing.T) {
 	})
 }
 
+// TestBlobFetcherPartialUnrecoverable checks that a partial fetch is discarded
+// once the remaining announcing peers can no longer cover the required custody
+func TestBlobFetcherPartialUnrecoverable(t *testing.T) {
+	testBlobFetcher(t, blobFetcherTest{
+		init: func() *BlobFetcher {
+			return NewBlobFetcher(
+				BlobFetcherFunctions{
+					HasPayload:    func(common.Hash) bool { return false },
+					AddCells:      func(common.Hash, map[string]*PeerCellDelivery, types.CustodyBitmap) {},
+					FetchPayloads: func(string, []common.Hash, types.CustodyBitmap) error { return nil },
+					DropPeer:      func(string) {},
+				},
+				custody,
+				&mockRand{value: 60}, // Force partial requests
+				15,
+			)
+		},
+		steps: []interface{}{
+			// Two full announces pass the availability check and start a partial fetch (A requested).
+			doBlobNotify{peer: "A", hashes: []common.Hash{testBlobTxHashes[0]}, custody: fullCustody},
+			doBlobNotify{peer: "C", hashes: []common.Hash{testBlobTxHashes[0]}, custody: fullCustody},
+			isFetching{hashes: map[common.Hash]fetchInfo{
+				testBlobTxHashes[0]: {fetching: &custody, fetched: []uint64{}},
+			}},
+
+			// Drop A: C (full) takes over the whole in-flight custody.
+			doDrop("A"),
+			isBlobScheduled{
+				announces: map[string][]blobAnnounce{
+					"C": {{hash: testBlobTxHashes[0], custody: custody}},
+				},
+				fetching: map[string][]blobAnnounce{
+					"C": {{hash: testBlobTxHashes[0], custody: custody}},
+				},
+			},
+
+			// A partial peer announces part of the custody. C already has all of it in
+			// flight, so B is only recorded as an alternate (no request of its own).
+			doBlobNotify{peer: "B", hashes: []common.Hash{testBlobTxHashes[0]}, custody: frontCustody},
+			isBlobScheduled{
+				announces: map[string][]blobAnnounce{
+					"C": {{hash: testBlobTxHashes[0], custody: custody}},
+					"B": {{hash: testBlobTxHashes[0], custody: frontCustody.Intersection(custody)}},
+				},
+				fetching: map[string][]blobAnnounce{
+					"C": {{hash: testBlobTxHashes[0], custody: custody}},
+				},
+			},
+
+			// Drop C: only B's partial custody remains, which cannot cover the
+			// required custody. The tx is unrecoverable, so every trace of it is
+			// discarded, including B's lingering announce.
+			doDrop("C"),
+			isFetching{hashes: nil},
+			isBlobScheduled{
+				announces: nil,
+				fetching:  nil,
+			},
+		},
+	})
+}
+
 // TestBlobFetcherFetchTimeout tests fetch timeout and rescheduling, full request case
 func TestBlobFetcherFetchTimeout(t *testing.T) {
 	testBlobFetcher(t, blobFetcherTest{