eth/fetcher: record latency samples only for accepted deliveries

A latency success sample was recorded for any in-time direct delivery,
regardless of content. That made samples free to farm: announce junk,
answer the fetch fast with duplicates or rejects, and accumulate the
sample count and freshness the dropper's latency protection requires.

Gate the success sample on the delivery containing at least one tx that
the pool newly accepted, reusing the acceptance set already computed for
onAccepted. Earning a latency sample now requires the same thing the
dropper wants to reward: sourcing novel valid transactions. Duplicate
deliveries are already-known to the pool, so a Sybil cluster replaying
one tx stream across identities cannot farm samples either — only the
first deliverer's fetch counts.

Timeout samples are unchanged: penalties must not be gateable.

eth/fetcher/tx_fetcher.go

@@ -127,6 +127,7 @@ type txDelivery struct {
 	hashes    []common.Hash // Batch of transaction hashes having been delivered
 	metas     []txMetadata  // Batch of metadata associated with the delivered hashes
 	direct    bool          // Whether this is a direct reply or a broadcast
+	accepted  bool          // Whether any delivered tx was newly accepted by the pool
 	violation error         // Whether we encountered a protocol violation
 }
 
@@ -188,7 +189,7 @@ type TxFetcher struct {
 	fetchTxs        func(string, []common.Hash) error                      // Retrieves a set of txs from a remote peer
 	dropPeer        func(string)                                           // Drops a peer in case of announcement violation
 	onAccepted      func(peer string, hashes []common.Hash)                // Optional: notified with accepted tx hashes per peer
-	onRequestResult func(peer string, latency time.Duration, timeout bool) // Optional: notified once per completed/timed-out tx request
+	onRequestResult func(peer string, latency time.Duration, timeout bool) // Optional: notified per timed-out request and per delivery with pool-accepted txs
 
 	buffer *blobpool.BlobBuffer
 
@@ -357,8 +358,9 @@ func (f *TxFetcher) Enqueue(peer string, version uint, txs []*types.Transaction,
 	// Push all the transactions into the pool, tracking underpriced ones to avoid
 	// re-requesting them and dropping the peer in case of malicious transfers.
 	var (
-		added = make([]common.Hash, 0, len(txs))
-		metas = make([]txMetadata, 0, len(txs))
+		added       = make([]common.Hash, 0, len(txs))
+		metas       = make([]txMetadata, 0, len(txs))
+		anyAccepted bool
 	)
 	// proceed in batches
 	for i := 0; i < len(txs); i += addTxsBatchSize {
@@ -416,8 +418,11 @@ func (f *TxFetcher) Enqueue(peer string, version uint, txs []*types.Transaction,
 		otherreject := f.handleAddErrors(hashes, errs, metrics)
 
 		// Notify the tracker which txs from this peer were accepted.
-		if f.onAccepted != nil && len(accepted) > 0 {
-			f.onAccepted(peer, accepted)
+		if len(accepted) > 0 {
+			anyAccepted = true
+			if f.onAccepted != nil {
+				f.onAccepted(peer, accepted)
+			}
 		}
 		// If 'other reject' is >25% of the deliveries in any batch, sleep a bit
 		// to throttle the misbehaving peer.
@@ -431,7 +436,7 @@ func (f *TxFetcher) Enqueue(peer string, version uint, txs []*types.Transaction,
 		}
 	}
 	select {
-	case f.cleanup <- &txDelivery{origin: peer, hashes: added, metas: metas, direct: direct, violation: violation}:
+	case f.cleanup <- &txDelivery{origin: peer, hashes: added, metas: metas, direct: direct, accepted: anyAccepted, violation: violation}:
 		return nil
 	case <-f.quit:
 		return errTerminated
@@ -843,8 +848,11 @@ func (f *TxFetcher) loop() {
 					txFetcherSlowWait.Update(time.Duration(f.clock.Now() - req.time).Nanoseconds())
 					// Already counted as a timeout sample at the timeout site;
 					// don't double-record on eventual delivery.
-				} else if f.onRequestResult != nil {
-					// Normal in-time delivery. Record the actual round-trip.
+				} else if f.onRequestResult != nil && delivery.accepted {
+					// In-time delivery that yielded at least one pool-accepted
+					// tx. Record the actual round-trip. Deliveries without any
+					// accepted tx (duplicates, rejects) record nothing — latency
+					// samples must not be farmable from worthless responses.
 					f.onRequestResult(delivery.origin, time.Duration(f.clock.Now()-req.time), false)
 				}
 				delete(f.requests, delivery.origin)

eth/fetcher/tx_fetcher_test.go

@@ -2403,3 +2403,41 @@ func TestTransactionFetcherRequestResultOnTimeout(t *testing.T) {
 		},
 	})
 }
+
+// TestTransactionFetcherRequestResultRequiresAcceptance asserts that an
+// in-time delivery containing no pool-accepted transactions (e.g. all
+// duplicates) does NOT record a latency sample — a peer cannot farm
+// latency protection by answering fetches with worthless content.
+func TestTransactionFetcherRequestResultRequiresAcceptance(t *testing.T) {
+	rec := &resultRecorder{}
+	testTransactionFetcherParallel(t, txFetcherTest{
+		init: func() *TxFetcher {
+			f := NewTxFetcher(
+				nil,
+				func(common.Hash, byte) error { return nil },
+				func(txs []*types.Transaction) []error {
+					errs := make([]error, len(txs))
+					for i := range errs {
+						errs[i] = txpool.ErrAlreadyKnown
+					}
+					return errs
+				},
+				func(string, []common.Hash) error { return nil },
+				nil, nil, rec.record,
+				newTestBlobBuffer(),
+			)
+			return f
+		},
+		steps: []interface{}{
+			doTxNotify{peer: "A", hashes: []common.Hash{testTxsHashes[0]}, types: []byte{testTxs[0].Type()}, sizes: []uint32{uint32(testTxs[0].Size())}},
+			doWait{time: txArriveTimeout, step: true},
+			doWait{time: 200 * time.Millisecond, step: false},
+			doTxEnqueue{peer: "A", txs: []*types.Transaction{testTxs[0]}, direct: true},
+			doFunc(func() {
+				if samples := rec.snapshot(); len(samples) != 0 {
+					t.Fatalf("expected no sample for unaccepted delivery, got %d (%v)", len(samples), samples)
+				}
+			}),
+		},
+	})
+}