diff --git a/core/txpool/legacypool/legacypool.go b/core/txpool/legacypool/legacypool.go index 062e074092f..74c35466360 100644 --- a/core/txpool/legacypool/legacypool.go +++ b/core/txpool/legacypool/legacypool.go @@ -244,6 +244,7 @@ type LegacyPool struct { chain BlockChain gasTip atomic.Pointer[uint256.Int] txFeed event.Feed + scope event.SubscriptionScope // Subscription scope to unsubscribe all on shutdown signer types.Signer mu sync.RWMutex @@ -293,7 +294,7 @@ func New(config Config, chain BlockChain) *LegacyPool { all: newLookup(), reqResetCh: make(chan *txpoolResetRequest), reqPromoteCh: make(chan *accountSet), - queueTxEventCh: make(chan *types.Transaction), + queueTxEventCh: make(chan *types.Transaction, 1), reorgDoneCh: make(chan chan struct{}), reorgShutdownCh: make(chan struct{}), initDoneCh: make(chan struct{}), @@ -393,6 +394,10 @@ func (pool *LegacyPool) loop() { func (pool *LegacyPool) Close() error { // Terminate the pool reorger and return close(pool.reorgShutdownCh) + // Unsubscribe anyone still listening for tx events. This also wakes up a + // runReorg that may be blocked in txFeed.Send because a subscriber stopped + // draining, allowing scheduleReorgLoop to observe the shutdown and return. + pool.scope.Close() pool.wg.Wait() log.Info("Transaction pool stopped") @@ -413,7 +418,7 @@ func (pool *LegacyPool) SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs // hard to separate newly discovered transactions from resurrected ones. This // is because the new txs are added to the queue, resurrected ones too and // reorgs run lazily, so separating the two would need a marker. - return pool.txFeed.Subscribe(ch) + return txpool.TrackOrTerminated(&pool.scope, pool.txFeed.Subscribe(ch)) } // SetGasTip updates the minimum gas tip required by the transaction pool for a @@ -986,7 +991,7 @@ func (pool *LegacyPool) promoteSpecialTx(addr common.Address, tx *types.Transact // Set the potentially new pending nonce and notify any subsystems of the new tx pool.queue.bump(addr) pool.pendingNonces.set(addr, tx.Nonce()+1) - pool.txFeed.Send(core.NewTxsEvent{Txs: []*types.Transaction{tx}}) + pool.queueTxEvent(tx) return true, nil } @@ -1221,6 +1226,10 @@ func (pool *LegacyPool) requestPromoteExecutables(set *accountSet) chan struct{} } // queueTxEvent enqueues a transaction event to be sent in the next reorg run. +// The channel is size-1 so callers that enqueue while holding the pool write +// lock (e.g. promoteSpecialTx) are decoupled from scheduleReorgLoop's select +// loop and never block past the first pending event; during shutdown the send +// falls through to reorgShutdownCh and returns immediately. func (pool *LegacyPool) queueTxEvent(tx *types.Transaction) { select { case pool.queueTxEventCh <- tx: @@ -1276,16 +1285,26 @@ func (pool *LegacyPool) scheduleReorgLoop() { pool.reorgDoneCh <- nextDone case tx := <-pool.queueTxEventCh: - // Queue up the event, but don't schedule a reorg. It's up to the caller to - // request one later if they want the events sent. + // Queue up the event, but don't schedule a reorg unless the pool is + // idle: callers that queued an event without requesting a reorg + // (e.g. a special tx promoted straight to pending) still expect it + // delivered, so schedule a run if none is running or pending. addr, _ := types.Sender(pool.signer, tx) if _, ok := queuedEvents[addr]; !ok { queuedEvents[addr] = NewSortedMap() } queuedEvents[addr].Put(tx) + if curDone == nil && !launchNextRun { + launchNextRun = true + } case <-curDone: curDone = nil + // Deliver any events queued while the run was active: schedule one + // more run so they don't wait indefinitely on an idle chain. + if len(queuedEvents) > 0 && !launchNextRun { + launchNextRun = true + } case <-pool.reorgShutdownCh: // Wait for current run to finish. diff --git a/core/txpool/legacypool/legacypool_test.go b/core/txpool/legacypool/legacypool_test.go index c907a97f73f..3b1c1161de7 100644 --- a/core/txpool/legacypool/legacypool_test.go +++ b/core/txpool/legacypool/legacypool_test.go @@ -3670,3 +3670,187 @@ func TestSetGasPrice(t *testing.T) { }) } } + +// TestSpecialTxPromotionDoesNotBlockOnTxFeed reproduces the mainnet freeze: promoting a +// special transaction delivered its NewTxsEvent while holding the pool write lock, so a +// subscriber that stopped draining wedged the pool itself and, through it, every peer +// goroutine that wanted to add or read transactions. +func TestSpecialTxPromotionDoesNotBlockOnTxFeed(t *testing.T) { + pool, key := setupPool() + defer pool.Close() + + pool.SetSigner(func(common.Address) bool { return true }) + testAddBalance(pool, crypto.PubkeyToAddress(key.PublicKey), big.NewInt(1_000_000_000_000_000_000)) + + // Subscriber that never reads, modelling a stalled txBroadcastLoop. + sink := make(chan core.NewTxsEvent) + sub := pool.SubscribeTransactions(sink, false) + defer sub.Unsubscribe() + + gasPrice := new(big.Int).SetUint64(common.DefaultMinGasPrice + 1) + specialTx, err := types.SignTx(types.NewTransaction(0, common.BlockSignersBinary, big.NewInt(1), 100000, gasPrice, nil), types.HomesteadSigner{}, key) + if err != nil { + t.Fatalf("failed to sign special tx: %v", err) + } + if !specialTx.IsSpecialTransaction() { + t.Fatal("test setup: transaction is not special") + } + + added := make(chan error, 1) + go func() { + added <- pool.Add([]*types.Transaction{specialTx}, false)[0] + }() + select { + case err := <-added: + if err != nil { + t.Fatalf("failed to add special tx: %v", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Add blocked: the special tx event is delivered while holding the pool lock") + } + + // Rigidity check: the special tx must actually be promoted to pending, not just + // accepted into the queue. A regression that skipped promotion would let this test + // pass trivially (Add returns, the lock is free) while the tx never reached pending. + promoted := false + deadline := time.Now().Add(2 * time.Second) + for !promoted && time.Now().Before(deadline) { + pending, _ := pool.Content() + for _, txs := range pending { + for _, ptx := range txs { + if ptx.Hash() == specialTx.Hash() { + promoted = true + break + } + } + if promoted { + break + } + } + if !promoted { + time.Sleep(10 * time.Millisecond) + } + } + if !promoted { + t.Fatal("special tx was accepted but never promoted to pending") + } + + usable := make(chan struct{}) + go func() { + defer close(usable) + pool.Stats() + }() + select { + case <-usable: + case <-time.After(5 * time.Second): + t.Fatal("pool lock still held after promoting a special tx") + } +} + +// TestLegacyPoolCloseUnblocksStalledSubscriber locks in the L1 guarantee: a subscriber +// that stops draining its channel leaves runReorg blocked in txFeed.Send, so close(done) +// never fires and LegacyPool.Close (wg.Wait) would hang forever, forcing a hard SIGKILL. +// Closing the subscription scope must remove the stuck subscriber and let Send return, so +// Close completes without hanging. +func TestLegacyPoolCloseUnblocksStalledSubscriber(t *testing.T) { + pool, key := setupPool() + + // LegacyPool.Close is not idempotent (close(reorgShutdownCh) panics on the + // second call), so guard it: the deferred teardown closes the pool on any + // early failure path, while the body's goroutine uses the same Once. + var closeOnce sync.Once + defer closeOnce.Do(func() { pool.Close() }) + + pool.SetSigner(func(common.Address) bool { return true }) + testAddBalance(pool, crypto.PubkeyToAddress(key.PublicKey), big.NewInt(1_000_000_000_000_000_000)) + + // Stalled subscriber: never read, never explicitly unsubscribe (Close does it via + // the subscription scope). + sink := make(chan core.NewTxsEvent) + pool.SubscribeTransactions(sink, false) + + gasPrice := new(big.Int).SetUint64(common.DefaultMinGasPrice + 1) + tx, err := types.SignTx(types.NewTransaction(0, common.Address{0x01}, big.NewInt(1), 21000, gasPrice, nil), types.HomesteadSigner{}, key) + if err != nil { + t.Fatalf("failed to sign tx: %v", err) + } + + // A sync Add waits for the reorg to finish, which requires txFeed.Send to return. + // With a stalled subscriber Send never returns, so Add must block here: this proves + // the reorg is wedged (the pre-L1 behaviour), which is exactly the state Close must + // survive. + added := make(chan error, 1) + go func() { + added <- pool.Add([]*types.Transaction{tx}, true)[0] + }() + select { + case err := <-added: + // An early return here means either the reorg was not wedged (the fix + // already worked) or the tx was rejected for an unrelated reason, so + // surface the actual error instead of assuming the wedge. + t.Fatalf("sync Add returned although the subscriber is stalled (expected Send to block, err=%v)", err) + case <-time.After(2 * time.Second): + // Confirmed: runReorg is stuck in txFeed.Send. + } + + // Close must not hang: scope.Close() removes the stalled subscriber, Send returns, + // runReorg finishes and the reorg loop shuts down. + closed := make(chan struct{}) + go func() { + closeOnce.Do(func() { pool.Close() }) + close(closed) + }() + select { + case <-closed: + // Close returned as expected. + case <-time.After(5 * time.Second): + t.Fatal("Close hung: stalled subscriber blocked reorg shutdown (L1 regression)") + } + + // The previously blocked Add must now complete, because Send was unblocked by Close. + select { + case err := <-added: + if err != nil { + t.Fatalf("Add failed after Close unblocked Send: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Add did not complete after Close unblocked Send") + } +} + +// TestQueueTxEventDeliveredWhenIdle locks in the delivery guarantee for events +// queued without a reorg request: on an idle pool (no reset or promote pending) +// scheduleReorgLoop must schedule a run to deliver them, otherwise a special tx +// promoted straight to pending would never be announced. +func TestQueueTxEventDeliveredWhenIdle(t *testing.T) { + pool, key := setupPool() + defer pool.Close() + + pool.SetSigner(func(common.Address) bool { return true }) + testAddBalance(pool, crypto.PubkeyToAddress(key.PublicKey), big.NewInt(1_000_000_000_000_000_000)) + + // Draining subscriber. + events := make(chan core.NewTxsEvent, 1) + sub := pool.SubscribeTransactions(events, false) + defer sub.Unsubscribe() + + gasPrice := new(big.Int).SetUint64(common.DefaultMinGasPrice + 1) + tx, err := types.SignTx(types.NewTransaction(0, common.Address{0x01}, big.NewInt(1), 21000, gasPrice, nil), types.HomesteadSigner{}, key) + if err != nil { + t.Fatalf("failed to sign tx: %v", err) + } + + // Queue the event without any reorg request: the idle pool must still + // deliver it. Before the idle-launch fix this never happened, because + // scheduleReorgLoop only queued the event and no run was scheduled. + pool.queueTxEvent(tx) + + select { + case ev := <-events: + if len(ev.Txs) != 1 || ev.Txs[0].Hash() != tx.Hash() { + t.Fatalf("unexpected event: got %d txs, first hash %v want %v", len(ev.Txs), ev.Txs[0].Hash(), tx.Hash()) + } + case <-time.After(3 * time.Second): + t.Fatal("queued tx event was not delivered on an idle pool") + } +} diff --git a/core/txpool/subpool.go b/core/txpool/subpool.go index 2e2bf2b51c5..7d20cfb314c 100644 --- a/core/txpool/subpool.go +++ b/core/txpool/subpool.go @@ -135,7 +135,10 @@ type SubPool interface { // SubscribeTransactions subscribes to new transaction events. The subscriber // can decide whether to receive notifications only for newly seen transactions - // or also for reorged out ones. + // or also for reorged out ones. Implementations must never return a nil + // subscription: a pool that is already shutting down yields a terminated + // (already unsubscribed) subscription instead, so callers can always wait on + // Err() and call Unsubscribe(). SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs bool) event.Subscription // Nonce returns the next nonce of an account, with all transactions executable diff --git a/core/txpool/txpool.go b/core/txpool/txpool.go index 7c9943d13c2..6a39fd6ef1a 100644 --- a/core/txpool/txpool.go +++ b/core/txpool/txpool.go @@ -135,14 +135,19 @@ func (p *TxPool) Close() error { if err := <-errc; err != nil { errs = append(errs, err) } + // Unsubscribe anyone still listening for tx events. This must happen before + // terminating the subpools: a subscriber that stopped draining its channel + // would otherwise leave a subpool's runReorg blocked in txFeed.Send, and the + // subpool Close (wg.Wait) would hang indefinitely. Closing the scope removes + // the stuck subscription and unblocks Send. + p.subs.Close() + // Terminate each subpool for _, subpool := range p.subpools { if err := subpool.Close(); err != nil { errs = append(errs, err) } } - // Unsubscribe anyone still listening for tx events - p.subs.Close() if len(errs) > 0 { return fmt.Errorf("subpool close errors: %v", errs) @@ -390,14 +395,31 @@ func (p *TxPool) Pending(filter PendingFilter) map[common.Address][]*LazyTransac return txs } +// TrackOrTerminated tracks sub in scope and returns it. If the scope is +// already closed (pool shutting down), Track refuses and returns nil; in +// that case sub is unsubscribed and a terminated subscription is returned +// instead, so callers can always safely wait on Err() and call Unsubscribe() +// without leaking the refused subscription. +func TrackOrTerminated(scope *event.SubscriptionScope, sub event.Subscription) event.Subscription { + if tracked := scope.Track(sub); tracked != nil { + return tracked + } + sub.Unsubscribe() + return event.NewSubscription(func(quit <-chan struct{}) error { return nil }) +} + // SubscribeTransactions registers a subscription for new transaction events, // supporting feeding only newly seen or also resurrected transactions. func (p *TxPool) SubscribeTransactions(ch chan<- core.NewTxsEvent, reorgs bool) event.Subscription { subs := make([]event.Subscription, len(p.subpools)) for i, subpool := range p.subpools { + // TrackOrTerminated never returns nil: a subpool that is already + // shutting down yields a terminated subscription instead, keeping + // JoinSubscriptions from panicking on a nil entry when the joined + // subscription terminates. subs[i] = subpool.SubscribeTransactions(ch, reorgs) } - return p.subs.Track(event.JoinSubscriptions(subs...)) + return TrackOrTerminated(&p.subs, event.JoinSubscriptions(subs...)) } // PoolNonce returns the next nonce of an account, with all transactions executable diff --git a/eth/handler.go b/eth/handler.go index 1d2b63c1793..0472b345b9c 100644 --- a/eth/handler.go +++ b/eth/handler.go @@ -1043,7 +1043,7 @@ func (pm *ProtocolManager) BroadcastTransactions(txs types.Transactions, propaga log.Trace("Broadcast transaction", "hash", tx.Hash(), "recipients", len(peers)) } for peer, hashes := range txset { - peer.AsyncSendTransactions(hashes) + peer.asyncSendChunked(hashes, false) } return } @@ -1055,11 +1055,7 @@ func (pm *ProtocolManager) BroadcastTransactions(txs types.Transactions, propaga } } for peer, hashes := range annos { - if peer.version >= xdc165 { - peer.AsyncSendPooledTransactionHashes(hashes) - } else { - peer.AsyncSendTransactions(hashes) - } + peer.asyncSendChunked(hashes, peer.version >= xdc165) } } diff --git a/eth/peer.go b/eth/peer.go index daf97d0b23e..6ad5ca42248 100644 --- a/eth/peer.go +++ b/eth/peer.go @@ -26,6 +26,7 @@ import ( "github.com/XinFinOrg/XDPoSChain/common" "github.com/XinFinOrg/XDPoSChain/core/forkid" "github.com/XinFinOrg/XDPoSChain/core/types" + "github.com/XinFinOrg/XDPoSChain/metrics" "github.com/XinFinOrg/XDPoSChain/p2p" "github.com/XinFinOrg/XDPoSChain/rlp" mapset "github.com/deckarep/golang-set/v2" @@ -51,6 +52,27 @@ const ( // before dropping older announcements. maxQueuedTxAnns = 4096 + // txBatchBuffer is the number of transaction batches (propagations or + // announcements) that may be queued for a peer before newer batches are + // dropped. Together with the non-blocking send in the AsyncSend* methods, + // it guards the shared broadcast loop against a stalled or exited per-peer + // writer blocking it indefinitely. Batches are limited to maxTxBatchSize + // hashes (see asyncSendChunked), so the channel itself holds at most + // txBatchBuffer * maxTxBatchSize = 4096 hashes, matching maxQueuedTxs and + // maxQueuedTxAnns. The true per-peer in-flight total is up to 8192 per + // channel: 4096 buffered plus another 4096 in the writer's own capped + // queue (which accumulates dequeued batches, keeping the newest), and + // xdc/165+ peers run both the broadcast and the announce channel, so up + // to 16384 hashes can be marked known but unsent per peer. + txBatchBuffer = 16 + + // maxTxBatchSize bounds the number of transaction hashes enqueued in a + // single AsyncSend* call. Callers may pass arbitrarily large sets (e.g. + // the whole pending pool during initial sync); chunking them keeps the + // per-peer queue depth bounded by batch count times batch size instead of + // by the size of the entire transaction set. + maxTxBatchSize = 256 + // maxQueuedBlocks is the maximum number of block propagations to queue up before // dropping broadcasts. There's not much point in queueing stale blocks, so a few // that might cover uncles should be enough. @@ -114,6 +136,27 @@ type peer struct { knownSyncInfo mapset.Set[common.Hash] // Set of BFT Sync Info known to be known by this peer } +var ( + // Note on divergence from upstream geth: upstream's AsyncSendTransactions / + // AsyncSendPooledTransactionHashes block on the (unbuffered) per-peer channel + // until the writer accepts the batch, selecting on p.term for shutdown. This + // fork deliberately diverges: sends are non-blocking with drop-on-full, so a + // stalled or exited peer writer can never stall the node-wide broadcast loop + // (upstream's writer loop exits permanently on network error, and p.term is + // only closed by the test-only peer.close). Consequence: transaction + // propagation/announcement to a peer is best-effort — under queue pressure + // the batch is dropped and the peer misses those transactions until + // reconnect. These meters are the only signal that drops are happening. + // + // txBroadcastDropMeter measures the number of transaction hashes dropped + // from propagation because the peer's broadcast queue was full. + txBroadcastDropMeter = metrics.NewRegisteredMeter("eth/peer/transaction/broadcasts/drop", nil) + + // txAnnounceDropMeter measures the number of transaction hashes dropped + // from announcement because the peer's announce queue was full. + txAnnounceDropMeter = metrics.NewRegisteredMeter("eth/peer/transaction/announces/drop", nil) +) + func newPeer(version int, p *p2p.Peer, rw p2p.MsgReadWriter, getPooledTx func(hash common.Hash) *types.Transaction) *peer { return &peer{ Peer: p, @@ -126,8 +169,8 @@ func newPeer(version int, p *p2p.Peer, rw p2p.MsgReadWriter, getPooledTx func(ha knownLendingTxs: mapset.NewSet[common.Hash](), queuedBlocks: make(chan *propEvent, maxQueuedBlocks), queuedBlockAnns: make(chan *types.Block, maxQueuedBlockAnns), - txBroadcast: make(chan []common.Hash), - txAnnounce: make(chan []common.Hash), + txBroadcast: make(chan []common.Hash, txBatchBuffer), + txAnnounce: make(chan []common.Hash, txBatchBuffer), getPooledTx: getPooledTx, term: make(chan struct{}), @@ -443,9 +486,21 @@ func (p *peer) SendLendingTransactions(txs types.LendingTransactions) error { } // AsyncSendTransactions queues a list of transactions (by hash) to eventually -// propagate to a remote peer. The number of pending sends are capped (new ones -// will force old sends to be dropped) +// propagate to a remote peer. The send never blocks: if the peer's broadcast +// queue is full or the writer is stalled, the batch is dropped. Dropped +// batches are not retried. Their hashes are left unmarked (not known), so the +// peer stays eligible if the same transactions are broadcast again later, but +// the txpool emits events only for newly added transactions, so in practice +// the peer catches up on reconnect, when the pending pool is re-sent. func (p *peer) AsyncSendTransactions(hashes []common.Hash) { + // Check termination first so a shutdown-period drop is attributed to the + // peer closing rather than a full queue, keeping the drop meters accurate. + select { + case <-p.term: + p.Log().Debug("Dropping transaction propagation", "count", len(hashes)) + return + default: + } select { case p.txBroadcast <- hashes: // Mark all the transactions as known, but ensure we don't overflow our limits @@ -457,6 +512,9 @@ func (p *peer) AsyncSendTransactions(hashes []common.Hash) { } case <-p.term: p.Log().Debug("Dropping transaction propagation", "count", len(hashes)) + default: + txBroadcastDropMeter.Mark(int64(len(hashes))) + p.Log().Debug("Dropping transaction propagation (queue full)", "count", len(hashes)) } } @@ -478,9 +536,21 @@ func (p *peer) sendPooledTransactionHashes(hashes []common.Hash) error { } // AsyncSendPooledTransactionHashes queues a list of transactions hashes to eventually -// announce to a remote peer. The number of pending sends are capped (new ones -// will force old sends to be dropped) +// announce to a remote peer. The send never blocks: if the peer's announce queue +// is full or the writer is stalled, the batch is dropped. Dropped batches are +// not retried; their hashes are left unmarked (not known), so the peer stays +// eligible if the same transactions are re-announced, but the txpool emits +// events only for newly added transactions, so in practice the peer catches up +// on reconnect. func (p *peer) AsyncSendPooledTransactionHashes(hashes []common.Hash) { + // Check termination first so a shutdown-period drop is attributed to the + // peer closing rather than a full queue, keeping the drop meters accurate. + select { + case <-p.term: + p.Log().Debug("Dropping transaction announcement", "count", len(hashes)) + return + default: + } select { case p.txAnnounce <- hashes: // Mark all the transactions as known, but ensure we don't overflow our limits @@ -492,6 +562,25 @@ func (p *peer) AsyncSendPooledTransactionHashes(hashes []common.Hash) { } case <-p.term: p.Log().Debug("Dropping transaction announcement", "count", len(hashes)) + default: + txAnnounceDropMeter.Mark(int64(len(hashes))) + p.Log().Debug("Dropping transaction announcement (queue full)", "count", len(hashes)) + } +} + +// asyncSendChunked splits an arbitrarily large hash set into maxTxBatchSize +// chunks and enqueues each via the AsyncSend* methods, keeping the per-peer +// queue depth bounded (txBatchBuffer * maxTxBatchSize hashes) no matter how +// large the caller's set is. announce selects AsyncSendPooledTransactionHashes +// (xdc/165+ announcement), otherwise AsyncSendTransactions (full propagation). +func (p *peer) asyncSendChunked(hashes []common.Hash, announce bool) { + for i := 0; i < len(hashes); i += maxTxBatchSize { + chunk := hashes[i:min(i+maxTxBatchSize, len(hashes))] + if announce { + p.AsyncSendPooledTransactionHashes(chunk) + } else { + p.AsyncSendTransactions(chunk) + } } } diff --git a/eth/peer_test.go b/eth/peer_test.go index 95c43062db1..f3c43a0dfad 100644 --- a/eth/peer_test.go +++ b/eth/peer_test.go @@ -1,6 +1,13 @@ package eth -import "testing" +import ( + "testing" + "time" + + "github.com/XinFinOrg/XDPoSChain/common" + "github.com/XinFinOrg/XDPoSChain/p2p" + "github.com/XinFinOrg/XDPoSChain/p2p/enode" +) func TestPeerSetRegisterRejectsDuplicateID(t *testing.T) { peers := newPeerSet() @@ -20,3 +27,136 @@ func TestPeerSetRegisterRejectsDuplicateID(t *testing.T) { t.Fatalf("registered peer replaced: got %p want %p", got, first) } } + +func TestAsyncSendDropsOnFullQueue(t *testing.T) { + tests := []struct { + name string + send func(*peer, []common.Hash) + }{ + {"broadcast", func(p *peer, hashes []common.Hash) { p.AsyncSendTransactions(hashes) }}, + {"announce", func(p *peer, hashes []common.Hash) { p.AsyncSendPooledTransactionHashes(hashes) }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + p := newPeer(xdc164, p2p.NewPeer(enode.ID{1}, "", nil), nil, nil) + // Fill the queue to capacity; every enqueue succeeds and marks known. + for i := range txBatchBuffer { + tt.send(p, []common.Hash{{byte(i + 1)}}) + } + // The next batch must be dropped immediately, not block the caller. + dropped := []common.Hash{{0xff}} + done := make(chan struct{}) + go func() { + tt.send(p, dropped) + close(done) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("send blocked on full queue") + } + // Dropped batch must not be marked known, so the peer stays eligible + // for re-propagation (marking happens only on successful enqueue). + if p.knownTxs.Contains(dropped[0]) { + t.Fatal("dropped batch was marked known") + } + if got := p.knownTxs.Cardinality(); got != txBatchBuffer { + t.Fatalf("known set size mismatch: got %d want %d", got, txBatchBuffer) + } + }) + } +} + +func TestAsyncSendChunkedDropsOnFullQueue(t *testing.T) { + tests := []struct { + name string + announce bool + }{ + {"broadcast", false}, + {"announce", true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + p := newPeer(xdc164, p2p.NewPeer(enode.ID{1}, "", nil), nil, nil) + // Fill the queue to capacity with small batches. + for i := range txBatchBuffer { + p.asyncSendChunked([]common.Hash{{byte(i + 1)}}, tt.announce) + } + // A huge chunked send over an already-full queue must drop immediately + // (no chunk blocks) and must not grow the in-flight set past the + // txBatchBuffer batches that were enqueued. + big := make([]common.Hash, 2*maxTxBatchSize) + for i := range big { + big[i] = common.Hash{byte(0x80 + i%0x7f)} + } + done := make(chan struct{}) + go func() { + p.asyncSendChunked(big, tt.announce) + close(done) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("chunked send blocked on full queue") + } + if got := p.knownTxs.Cardinality(); got != txBatchBuffer { + t.Fatalf("known set size mismatch: got %d want %d", got, txBatchBuffer) + } + }) + } +} + +func TestAsyncSendChunkedPartialDelivery(t *testing.T) { + tests := []struct { + name string + announce bool + }{ + {"broadcast", false}, + {"announce", true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + p := newPeer(xdc164, p2p.NewPeer(enode.ID{1}, "", nil), nil, nil) + // Leave the queue half empty: a subsequent chunked send enqueues + // exactly the remaining slots and drops the rest, exercising the + // partial-delivery contract and the chunk-boundary math. + for i := range txBatchBuffer / 2 { + p.asyncSendChunked([]common.Hash{{byte(i + 1)}}, tt.announce) + } + // 9 full chunks + a 1-hash tail: the 8 empty slots take the first 8 + // chunks (8*maxTxBatchSize hashes), the last two chunks drop. + big := make([]common.Hash, 9*maxTxBatchSize+1) + for i := range big { + // Two-byte index encoding: (i%128, i>>7) is bijective for the + // range used here, so every hash is unique, and the 0x80 offset + // keeps them disjoint from the fill hashes {1..8,0,0,...}. + big[i] = common.Hash{byte(0x80 + i%128), byte(i >> 7)} + } + done := make(chan struct{}) + go func() { + p.asyncSendChunked(big, tt.announce) + close(done) + }() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("chunked send blocked on full queue") + } + // Enqueued chunks are marked known (boundary: last hash of chunk 8), + // dropped chunks are not (first hash of chunk 9, tail of chunk 10). + enqueued := 8 * maxTxBatchSize + if !p.knownTxs.Contains(big[enqueued-1]) { + t.Fatal("last enqueued chunk hash not marked known") + } + if p.knownTxs.Contains(big[enqueued]) { + t.Fatal("dropped chunk hash was marked known") + } + if p.knownTxs.Contains(big[len(big)-1]) { + t.Fatal("dropped tail chunk hash was marked known") + } + if got := p.knownTxs.Cardinality(); got != txBatchBuffer/2+enqueued { + t.Fatalf("known set size mismatch: got %d want %d", got, txBatchBuffer/2+enqueued) + } + }) + } +} diff --git a/eth/sync.go b/eth/sync.go index faa410d6491..a9196210215 100644 --- a/eth/sync.go +++ b/eth/sync.go @@ -76,7 +76,7 @@ func (pm *ProtocolManager) syncTransactions(p *peer) { for i, tx := range txs { hashes[i] = tx.Hash() } - p.AsyncSendPooledTransactionHashes(hashes) + p.asyncSendChunked(hashes, true) return } // Out of luck, peer is running legacy protocols, drop the txs over