mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-07-21 20:26:41 +00:00
eth/txtracker: gate credit on pre-slot delivery
Add a per-entry arrival timestamp at NotifyAccepted time and skip inclusion and finalization credit when the delivery was at or after the slot of the inclusion block. This prevents a peer that learned a tx from the just-propagated block (or any post-slot source) from harvesting credit it didn't earn through genuine pre-slot relay work. Does not defeat builder-feed peers that deliver within the feed's lead time before slot start; but that's also not fundamentally different from sending to the mempool in the first place. Signed-off-by: Csaba Kiraly <csaba.kiraly@gmail.com>
This commit is contained in:
parent
ccd5fced83
commit
905bc9d102
2 changed files with 85 additions and 12 deletions
|
|
@ -28,6 +28,7 @@ package txtracker
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"sync"
|
"sync"
|
||||||
|
"time"
|
||||||
|
|
||||||
"github.com/ethereum/go-ethereum/common"
|
"github.com/ethereum/go-ethereum/common"
|
||||||
"github.com/ethereum/go-ethereum/core"
|
"github.com/ethereum/go-ethereum/core"
|
||||||
|
|
@ -67,11 +68,20 @@ type peerStats struct {
|
||||||
recentIncluded float64
|
recentIncluded float64
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// txEntry tracks the deliverer of a transaction and when delivery was
|
||||||
|
// recorded. addedAt is unix seconds; it is compared against block.Time()
|
||||||
|
// to suppress credit for txs delivered at or after the slot of their
|
||||||
|
// inclusion block (e.g., re-broadcasts of just-mined txs).
|
||||||
|
type txEntry struct {
|
||||||
|
peer string
|
||||||
|
addedAt uint64
|
||||||
|
}
|
||||||
|
|
||||||
// Tracker records which peer delivered each transaction and credits peers
|
// Tracker records which peer delivered each transaction and credits peers
|
||||||
// when their transactions appear on chain.
|
// when their transactions appear on chain.
|
||||||
type Tracker struct {
|
type Tracker struct {
|
||||||
mu sync.Mutex
|
mu sync.Mutex
|
||||||
txs map[common.Hash]string // hash → deliverer peer ID
|
txs map[common.Hash]txEntry // hash → deliverer + arrival time
|
||||||
peers map[string]*peerStats
|
peers map[string]*peerStats
|
||||||
order []common.Hash // insertion order for LRU eviction
|
order []common.Hash // insertion order for LRU eviction
|
||||||
|
|
||||||
|
|
@ -83,16 +93,18 @@ type Tracker struct {
|
||||||
quit chan struct{}
|
quit chan struct{}
|
||||||
stopOnce sync.Once
|
stopOnce sync.Once
|
||||||
step chan struct{} // test sync: sent after each event is processed
|
step chan struct{} // test sync: sent after each event is processed
|
||||||
|
now func() uint64 // unix-seconds clock; overridable in tests
|
||||||
wg sync.WaitGroup
|
wg sync.WaitGroup
|
||||||
}
|
}
|
||||||
|
|
||||||
// New creates a new tracker.
|
// New creates a new tracker.
|
||||||
func New() *Tracker {
|
func New() *Tracker {
|
||||||
return &Tracker{
|
return &Tracker{
|
||||||
txs: make(map[common.Hash]string),
|
txs: make(map[common.Hash]txEntry),
|
||||||
peers: make(map[string]*peerStats),
|
peers: make(map[string]*peerStats),
|
||||||
quit: make(chan struct{}),
|
quit: make(chan struct{}),
|
||||||
step: make(chan struct{}, 1),
|
step: make(chan struct{}, 1),
|
||||||
|
now: func() uint64 { return uint64(time.Now().Unix()) },
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -136,11 +148,12 @@ func (t *Tracker) NotifyAccepted(peer string, hashes []common.Hash) {
|
||||||
t.mu.Lock()
|
t.mu.Lock()
|
||||||
defer t.mu.Unlock()
|
defer t.mu.Unlock()
|
||||||
|
|
||||||
|
addedAt := t.now()
|
||||||
for _, hash := range hashes {
|
for _, hash := range hashes {
|
||||||
if _, ok := t.txs[hash]; ok {
|
if _, ok := t.txs[hash]; ok {
|
||||||
continue // already tracked, keep first deliverer
|
continue // already tracked, keep first deliverer
|
||||||
}
|
}
|
||||||
t.txs[hash] = peer
|
t.txs[hash] = txEntry{peer: peer, addedAt: addedAt}
|
||||||
t.order = append(t.order, hash)
|
t.order = append(t.order, hash)
|
||||||
}
|
}
|
||||||
// Ensure the delivering peer has a stats entry.
|
// Ensure the delivering peer has a stats entry.
|
||||||
|
|
@ -206,11 +219,17 @@ func (t *Tracker) handleChainHead(ev core.ChainHeadEvent) {
|
||||||
defer t.mu.Unlock()
|
defer t.mu.Unlock()
|
||||||
|
|
||||||
// Count per-peer inclusions in this block for the inclusion EMA.
|
// Count per-peer inclusions in this block for the inclusion EMA.
|
||||||
|
// Skip txs whose delivery arrived at or after this block's slot — those
|
||||||
|
// are likely post-slot re-broadcasts of an already-mined tx, not genuine
|
||||||
|
// relay work.
|
||||||
|
blockTime := block.Time()
|
||||||
blockIncl := make(map[string]int)
|
blockIncl := make(map[string]int)
|
||||||
for _, tx := range block.Transactions() {
|
for _, tx := range block.Transactions() {
|
||||||
if peer := t.txs[tx.Hash()]; peer != "" {
|
entry, ok := t.txs[tx.Hash()]
|
||||||
blockIncl[peer]++
|
if !ok || entry.addedAt >= blockTime {
|
||||||
|
continue
|
||||||
}
|
}
|
||||||
|
blockIncl[entry.peer]++
|
||||||
}
|
}
|
||||||
// Accumulate per-peer finalization credits over the newly-finalized
|
// Accumulate per-peer finalization credits over the newly-finalized
|
||||||
// range (possibly zero blocks). Only counts peers still tracked.
|
// range (possibly zero blocks). Only counts peers still tracked.
|
||||||
|
|
@ -245,15 +264,16 @@ func (t *Tracker) collectFinalizationCredits() map[string]int {
|
||||||
if block == nil {
|
if block == nil {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
blockTime := block.Time()
|
||||||
for _, tx := range block.Transactions() {
|
for _, tx := range block.Transactions() {
|
||||||
peer := t.txs[tx.Hash()]
|
entry, ok := t.txs[tx.Hash()]
|
||||||
if peer == "" {
|
if !ok || entry.addedAt >= blockTime {
|
||||||
continue
|
continue // unknown, or post-slot re-broadcast
|
||||||
}
|
}
|
||||||
if _, ok := t.peers[peer]; !ok {
|
if _, ok := t.peers[entry.peer]; !ok {
|
||||||
continue // peer disconnected, skip credit
|
continue // peer disconnected, skip credit
|
||||||
}
|
}
|
||||||
credits[peer]++
|
credits[entry.peer]++
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if total := sumCounts(credits); total > 0 {
|
if total := sumCounts(credits); total > 0 {
|
||||||
|
|
|
||||||
|
|
@ -90,12 +90,20 @@ func (c *mockChain) addBlock(num uint64, txs []*types.Transaction) *types.Block
|
||||||
// for reorg tests). If canonical is true, the block becomes the canonical
|
// for reorg tests). If canonical is true, the block becomes the canonical
|
||||||
// block for that height (looked up by GetBlockByNumber).
|
// block for that height (looked up by GetBlockByNumber).
|
||||||
func (c *mockChain) addBlockAtHeight(num, salt uint64, txs []*types.Transaction, canonical bool) *types.Block {
|
func (c *mockChain) addBlockAtHeight(num, salt uint64, txs []*types.Transaction, canonical bool) *types.Block {
|
||||||
|
return c.addBlockAtHeightWithTime(num, salt, txs, canonical, uint64(time.Now().Unix()+3600))
|
||||||
|
}
|
||||||
|
|
||||||
|
// addBlockAtHeightWithTime is like addBlockAtHeight but takes an explicit
|
||||||
|
// block time. Used by the pre-slot gate test, which needs a block whose
|
||||||
|
// slot start is BEFORE the moment NotifyAccepted recorded its tx.
|
||||||
|
func (c *mockChain) addBlockAtHeightWithTime(num, salt uint64, txs []*types.Transaction, canonical bool, blockTime uint64) *types.Block {
|
||||||
c.mu.Lock()
|
c.mu.Lock()
|
||||||
defer c.mu.Unlock()
|
defer c.mu.Unlock()
|
||||||
// Mix salt into Extra so siblings at the same height get distinct hashes.
|
// Mix salt into Extra so siblings at the same height get distinct hashes.
|
||||||
header := &types.Header{
|
header := &types.Header{
|
||||||
Number: new(big.Int).SetUint64(num),
|
Number: new(big.Int).SetUint64(num),
|
||||||
Extra: big.NewInt(int64(salt)).Bytes(),
|
Extra: big.NewInt(int64(salt)).Bytes(),
|
||||||
|
Time: blockTime,
|
||||||
}
|
}
|
||||||
block := types.NewBlock(header, &types.Body{Transactions: txs}, nil, trie.NewListHasher())
|
block := types.NewBlock(header, &types.Body{Transactions: txs}, nil, trie.NewListHasher())
|
||||||
c.blocksByHash[block.Hash()] = block
|
c.blocksByHash[block.Hash()] = block
|
||||||
|
|
@ -189,8 +197,8 @@ func TestNotifyReceived(t *testing.T) {
|
||||||
t.Fatalf("expected order length 3, got %d", len(tr.order))
|
t.Fatalf("expected order length 3, got %d", len(tr.order))
|
||||||
}
|
}
|
||||||
for i, h := range hashes {
|
for i, h := range hashes {
|
||||||
if got := tr.txs[h]; got != "peerA" {
|
if got := tr.txs[h]; got.peer != "peerA" {
|
||||||
t.Fatalf("tx %d: expected deliverer=peerA, got %q", i, got)
|
t.Fatalf("tx %d: expected deliverer=peerA, got %q", i, got.peer)
|
||||||
}
|
}
|
||||||
if tr.order[i] != h {
|
if tr.order[i] != h {
|
||||||
t.Fatalf("order[%d] mismatch", i)
|
t.Fatalf("order[%d] mismatch", i)
|
||||||
|
|
@ -454,3 +462,48 @@ func TestRecentFinalizedDecays(t *testing.T) {
|
||||||
t.Fatalf("expected RecentFinalized to decay, got %f >= peak %f", after, peak)
|
t.Fatalf("expected RecentFinalized to decay, got %f >= peak %f", after, peak)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestPreSlotGate verifies that a tx delivered at or after the slot of its
|
||||||
|
// inclusion block earns no credit. This blocks the simple
|
||||||
|
// post-block-propagation re-broadcast attribution attack: a peer that
|
||||||
|
// learns a tx from the just-mined block and re-broadcasts it to our pool
|
||||||
|
// should not gain credit when that block is processed. The finalization
|
||||||
|
// path applies the same gate (entry.addedAt >= blockTime) and is exercised
|
||||||
|
// by the existing TestFinalization with the new clock semantics.
|
||||||
|
func TestPreSlotGate(t *testing.T) {
|
||||||
|
tr := New()
|
||||||
|
chain := newMockChain()
|
||||||
|
tr.Start(chain)
|
||||||
|
defer tr.Stop()
|
||||||
|
|
||||||
|
// Pin the tracker's clock so NotifyAccepted records a known addedAt.
|
||||||
|
const delivery = uint64(1_000_000)
|
||||||
|
tr.now = func() uint64 { return delivery }
|
||||||
|
|
||||||
|
// peerA delivers two txs at the same instant.
|
||||||
|
preTx := makeTx(1)
|
||||||
|
postTx := makeTx(2)
|
||||||
|
tr.NotifyAccepted("peerA", []common.Hash{preTx.Hash(), postTx.Hash()})
|
||||||
|
|
||||||
|
// Block 1: slot strictly AFTER delivery — pre-slot, credit allowed.
|
||||||
|
chain.addBlockAtHeightWithTime(1, 1, []*types.Transaction{preTx}, true, delivery+100)
|
||||||
|
chain.sendHead(1)
|
||||||
|
waitStep(t, tr)
|
||||||
|
|
||||||
|
preEMA := tr.GetAllPeerStats()["peerA"].RecentIncluded
|
||||||
|
if preEMA <= 0 {
|
||||||
|
t.Fatalf("expected RecentIncluded>0 after pre-slot delivery, got %f", preEMA)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Block 2: slot strictly BEFORE delivery — post-slot, must NOT credit.
|
||||||
|
chain.addBlockAtHeightWithTime(2, 2, []*types.Transaction{postTx}, true, delivery-1)
|
||||||
|
chain.sendHead(2)
|
||||||
|
waitStep(t, tr)
|
||||||
|
|
||||||
|
// With the gate, only EMA decay occurs (no contribution this block).
|
||||||
|
// Without the gate, RecentIncluded would have ticked up again.
|
||||||
|
after := tr.GetAllPeerStats()["peerA"].RecentIncluded
|
||||||
|
if after >= preEMA {
|
||||||
|
t.Fatalf("expected EMA to decay (no credit for post-slot tx), got %f >= preEMA %f", after, preEMA)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue