diff --git a/miner/worker_pow.go b/miner/worker_pow.go
index 7cae05422..67bef0523 100644
--- a/miner/worker_pow.go
+++ b/miner/worker_pow.go
@@ -17,6 +17,7 @@
package miner
import (
+ "context"
"math/big"
"sync"
"sync/atomic"
@@ -41,6 +42,12 @@ const (
// chainHeadChanSize is the size of channel listening to ChainHeadEvent.
chainHeadChanSize = 10
+ // chainSideChanSize is the size of channel listening to side-block events.
+ chainSideChanSize = 10
+
+ // staleThreshold is the maximum depth of an acceptable stale uncle.
+ staleThreshold = 7
+
// sealingLogAtDepth is the number of confirmations before a sealing log is emitted.
sealingLogAtDepth = 7
@@ -79,12 +86,26 @@ type powWorker struct {
startCh chan struct{}
exitCh chan struct{}
resubmitIntervalCh chan time.Duration
+ uncleRecommitCh chan struct{} // signals newWorkLoop to regenerate work when a fresh uncle arrives
// Subscriptions
txsCh chan core.NewTxsEvent
txsSub event.Subscription
chainHeadCh chan core.ChainHeadEvent
chainHeadSub event.Subscription
+ chainSideCh chan core.ChainSideEvent
+ chainSideSub event.Subscription
+
+ // Uncle candidates sourced from the fork-choice side-block feed. Accessed
+ // only from mainLoop (single goroutine), so they need no extra locking.
+ localUncles map[common.Hash]*types.Header
+ remoteUncles map[common.Hash]*types.Header
+
+ // Test hooks (nil in production) that let worker_test.go drive the sealing
+ // loops deterministically.
+ newTaskHook func(*task)
+ skipSealHook func(*task) bool
+ fullTaskHook func()
// State
running atomic.Bool
@@ -117,14 +138,19 @@ func newPowWorker(miner *Miner, chain *core.BlockChain, pool *txpool.TxPool, eng
startCh: make(chan struct{}, 1),
exitCh: make(chan struct{}),
resubmitIntervalCh: make(chan time.Duration),
+ uncleRecommitCh: make(chan struct{}, 1),
txsCh: make(chan core.NewTxsEvent, txChanSize),
chainHeadCh: make(chan core.ChainHeadEvent, chainHeadChanSize),
+ chainSideCh: make(chan core.ChainSideEvent, chainSideChanSize),
+ localUncles: make(map[common.Hash]*types.Header),
+ remoteUncles: make(map[common.Hash]*types.Header),
pendingTasks: make(map[common.Hash]*task),
unconfirmed: newUnconfirmedBlocks(chain, sealingLogAtDepth),
}
// Subscribe to events
w.txsSub = pool.SubscribeTransactions(w.txsCh, false)
w.chainHeadSub = chain.SubscribeChainHeadEvent(w.chainHeadCh)
+ w.chainSideSub = chain.SubscribeChainSideEvent(w.chainSideCh)
// Sanitize recommit interval
recommit := miner.config.Recommit
@@ -215,14 +241,27 @@ func (w *powWorker) newWorkLoop(recommit time.Duration) {
}
timer.Reset(recommit)
}
+ // clearPending drops pending tasks whose blocks are now buried deeper than
+ // staleThreshold below the given height, so the map does not grow unbounded.
+ clearPending := func(number uint64) {
+ w.pendingMu.Lock()
+ for h, t := range w.pendingTasks {
+ if t.block.NumberU64()+staleThreshold <= number {
+ delete(w.pendingTasks, h)
+ }
+ }
+ w.pendingMu.Unlock()
+ }
for {
select {
case <-w.startCh:
+ clearPending(w.chain.CurrentBlock().Number.Uint64())
timestamp = time.Now().Unix()
commit(commitInterruptNewHead)
case <-w.chainHeadCh:
+ clearPending(w.chain.CurrentBlock().Number.Uint64())
timestamp = time.Now().Unix()
commit(commitInterruptNewHead)
@@ -235,6 +274,15 @@ func (w *powWorker) newWorkLoop(recommit time.Duration) {
w.newTxs.Store(0)
}
+ case <-w.uncleRecommitCh:
+ // A fresh uncle arrived; regenerate work now so it can be folded into the
+ // block being sealed (higher uncle reward), mirroring core-geth's
+ // recommit-on-side-block instead of waiting for the next resubmit.
+ if w.isRunning() {
+ timestamp = time.Now().Unix()
+ commit(commitInterruptResubmit)
+ }
+
case interval := <-w.resubmitIntervalCh:
// Adjust resubmit interval explicitly by the user.
if interval < minRecommitInterval {
@@ -255,12 +303,31 @@ func (w *powWorker) mainLoop() {
defer w.wg.Done()
defer w.txsSub.Unsubscribe()
defer w.chainHeadSub.Unsubscribe()
+ defer w.chainSideSub.Unsubscribe()
+
+ cleanTicker := time.NewTicker(time.Second * 10)
+ defer cleanTicker.Stop()
for {
select {
case req := <-w.newWorkCh:
w.commitWork(req.interrupt, req.timestamp)
+ case ev := <-w.chainSideCh:
+ // A block lost the fork choice; keep its header as a possible uncle. If it's
+ // new and we're sealing, ask newWorkLoop to regenerate work so the uncle is
+ // folded in immediately (like core-geth) rather than at the next resubmit.
+ if w.collectUncle(ev.Header) && w.isRunning() {
+ select {
+ case w.uncleRecommitCh <- struct{}{}:
+ default: // a recommit is already pending
+ }
+ }
+
+ case <-cleanTicker.C:
+ // Purge uncle candidates that have grown too old to be included.
+ w.pruneStaleUncles()
+
case <-w.txsCh:
// New transactions arrived, mark them for the next sealing round.
if w.isRunning() {
@@ -274,6 +341,8 @@ func (w *powWorker) mainLoop() {
return
case <-w.chainHeadSub.Err():
return
+ case <-w.chainSideSub.Err():
+ return
}
}
}
@@ -298,6 +367,9 @@ func (w *powWorker) taskLoop() {
for {
select {
case t := <-w.taskCh:
+ if w.newTaskHook != nil {
+ w.newTaskHook(t)
+ }
// Reject duplicate sealing work due to resubmitting.
sealHash := w.engine.SealHash(t.block.Header())
if sealHash == prev {
@@ -308,6 +380,11 @@ func (w *powWorker) taskLoop() {
stopCh = make(chan struct{})
prev = sealHash
+ // Allow tests to skip the (slow) sealing step.
+ if w.skipSealHook != nil && w.skipSealHook(t) {
+ continue
+ }
+
w.pendingMu.Lock()
w.pendingTasks[sealHash] = t
w.pendingMu.Unlock()
@@ -389,7 +466,7 @@ func (w *powWorker) commitWork(interrupt *atomic.Int32, timestamp int64) {
}
// Construct the sealing task using the existing Miner methods.
- work, err := w.miner.prepareWork(&generateParams{
+ work, err := w.miner.prepareWork(context.Background(), &generateParams{
timestamp: uint64(timestamp),
coinbase: coinbase,
}, false)
@@ -399,25 +476,31 @@ func (w *powWorker) commitWork(interrupt *atomic.Int32, timestamp int64) {
}
// Fill transactions.
- err = w.miner.fillTransactions(interrupt, work)
+ err = w.miner.fillTransactions(context.Background(), interrupt, work)
if err != nil {
log.Warn("Block building interrupted", "err", err)
}
+ // Gather up to two valid uncles for the block being sealed.
+ uncles := w.gatherUncles(work.header)
+
// Assemble the block.
body := types.Body{
Transactions: work.txs,
+ Uncles: uncles,
}
- block, err := w.miner.engine.FinalizeAndAssemble(w.chain, work.header, work.state, &body, work.receipts)
- if err != nil {
- log.Error("Failed to finalize block for sealing", "err", err)
- return
+ block := core.AssembleBlock(w.miner.engine, w.chain, work.header, work.state, &body, work.receipts)
+
+ // Allow tests to observe the full sealing task before it is pushed.
+ if w.fullTaskHook != nil {
+ w.fullTaskHook()
}
+
// Calculate fees for logging.
fees := totalFees(block, work.receipts)
if w.isRunning() {
log.Info("Commit new sealing work", "number", block.Number(), "sealhash", w.engine.SealHash(block.Header()),
- "txs", work.tcount, "gas", block.GasUsed(), "fees", ethToFloat(fees), "elapsed", common.PrettyDuration(time.Since(time.Unix(timestamp, 0))))
+ "uncles", len(uncles), "txs", work.tcount, "gas", block.GasUsed(), "fees", ethToFloat(fees), "elapsed", common.PrettyDuration(time.Since(time.Unix(timestamp, 0))))
}
select {