From 0ff1fb856fdf7bf2cd356c5a5e7836f1be87f770 Mon Sep 17 00:00:00 2001 From: Csaba Kiraly Date: Wed, 15 Jul 2026 15:34:45 +0200 Subject: [PATCH] eth/fetcher: record latency samples only for accepted deliveries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 24 ++++++++++++++------- eth/fetcher/tx_fetcher_test.go | 38 ++++++++++++++++++++++++++++++++++ 2 files changed, 54 insertions(+), 8 deletions(-) diff --git a/eth/fetcher/tx_fetcher.go b/eth/fetcher/tx_fetcher.go index f1d7337b96..cd0ea065b3 100644 --- a/eth/fetcher/tx_fetcher.go +++ b/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) diff --git a/eth/fetcher/tx_fetcher_test.go b/eth/fetcher/tx_fetcher_test.go index 27ad3f340f..37d401779a 100644 --- a/eth/fetcher/tx_fetcher_test.go +++ b/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) + } + }), + }, + }) +}