// core/txpool/txpool.go
// TxPool 是多个特定类型交易池的聚合器，统一管理节点认为“值得关注”的所有交易。
// 交易在从网络接收到或本地提交时进入交易池；
// 在被纳入区块链或因资源限制被驱逐时离开交易池。
type TxPool struct {
	subpools []SubPool // 用于处理不同类型交易的子池列表
	chain    BlockChain
	signer   types.Signer

	stateLock sync.RWMutex   // 保护 state 实例的读写锁
	state     *state.StateDB // 区块链头部区块对应的当前状态数据库

	subs event.SubscriptionScope // 订阅作用域，用于在关闭时统一取消所有订阅
	quit chan chan error         // 退出通道，用于终止区块头更新器
	term chan struct{}           // 终止信号通道，用于检测交易池是否已关闭

	sync chan chan error // 用于测试/模拟的通道，可阻塞等待内部重置完成
}

// core/txpool/legacypool/legacypool.go
// Config是交易池的配置参数。
type Config struct {
    Locals    []common.Address // 本地优先地址
    NoLocals  bool             // 是否禁用本地交易特殊处理

    Journal   string           // 本地交易日志文件路径（持久化未上链交易）
    Rejournal time.Duration    // 重新写入日志的间隔

    PriceLimit uint64          // 最小 gas price 阈值（低于此值的交易会被拒绝）
    PriceBump  uint64          // 替换交易时要求的 gas price 最小涨幅（百分比，如 10 表示 10%）

    AccountSlots uint64       // 每个账户在“可执行交易池”中保留的最小槽数
    GlobalSlots  uint64       // 整个可执行交易池最大交易数
    AccountQueue uint64       // 每个账户在“排队交易池”中允许的最大交易数
    GlobalQueue  uint64       // 整个排队交易池最大交易数

    Lifetime time.Duration    // 交易在池中的最大存活时间（超时会被丢弃）
}

// core/txpool/txpool.go
// loop是交易池的主事件循环，等待并响应外部区块链事件以及各种报告和交易驱逐事件。
func (p *TxPool) loop(head *types.Header) {
	// Close the termination marker when the pool stops
	defer close(p.term)

	// 订阅链头事件以触发子池重置
	var (
		newHeadCh  = make(chan core.ChainHeadEvent)
		newHeadSub = p.chain.SubscribeChainHeadEvent(newHeadCh)
	)
    ...
}

select {
case event := <-newHeadCh:
    // 仅更新 newHead，不立即触发 reset
    newHead = event.Header
    // 等待下一轮循环判断是否需要 reset
}

if newHead != oldHead || resetForced {
    select {
    case resetBusy <- struct{}{}:
        // 更新 TxPool 的全局 state（非子池状态）
        if statedb, err := p.chain.StateAt(newHead.Root); err == nil {
            p.stateLock.Lock()
            p.state = statedb
            p.stateLock.Unlock()
        }

        // 异步触发所有子池 Reset
        go func(oldHead, newHead *types.Header) {
            for _, subpool := range p.subpools {
                subpool.Reset(oldHead, newHead)
            }
            resetDone <- newHead
        }(oldHead, newHead)

        resetForced = false
    default:
        // reset 已在运行，跳过
    }
}

// core/txpool/legacypool/legacypool.go
func (pool *LegacyPool) Reset(oldHead, newHead *types.Header) {
    wait := pool.requestReset(oldHead, newHead)
    <-wait
}

// requestReset请求将池重置到新的头部块。
// 当重置完成后，返回的通道将被关闭。
func (pool *LegacyPool) requestReset(oldHead *types.Header, newHead *types.Header) chan struct{} {
	select {
	case pool.reqResetCh <- &txpoolResetRequest{oldHead, newHead}:
		return <-pool.reorgDoneCh
	case <-pool.reorgShutdownCh:
		return pool.reorgShutdownCh
	}
}

func New(gasTip uint64, chain BlockChain, subpools []SubPool) (*TxPool, error) {
    ...
	for i, subpool := range subpools {
		if err := subpool.Init(gasTip, head, reserver.NewHandle(i)); err != nil {
			for j := i - 1; j >= 0; j-- {
				subpools[j].Close()
			}
			return nil, err
		}
	}
	go pool.loop(head)
	return pool, nil
}

func (pool *LegacyPool) Init() error {
    // ... 初始化
    go pool.scheduleReorgLoop() // ← 启动 reorg 调度循环
    return nil
}

func (pool *LegacyPool) scheduleReorgLoop() {
    for {
        select {
        case req := <-pool.reqResetCh:
            // 收集 reset 请求
            reset = &txpoolResetRequest{oldHead: req.oldHead, newHead: req.newHead}

        case ... // 其他事件（如新交易）

        default:
            if reset != nil || len(dirtyAccounts) > 0 {
                // 启动批量 reorg 处理
                go pool.runReorg(nextDone, reset, dirtyAccounts, queuedEvents)
                reset, dirtyAccounts, queuedEvents = nil, nil, nil
            }
            time.Sleep(100 * time.Millisecond) // 避免空转
        }
    }
}


func (pool *LegacyPool) runReorg(done chan struct{}, reset *txpoolResetRequest, dirtyAccounts *accountSet, events TxEventQueue) {
    defer func() { done <- struct{}{} }()

    if reset != nil {
        // 真正执行状态重置和交易池清理
        pool.reset(reset.oldHead, reset.newHead)
    }

    // ... 处理 dirtyAccounts、events（如 promoteExecutables）

    // 通知外部 reset 完成
    select {
    case pool.reorgDoneCh <- struct{}{}:
    default:
    }
}

// eth/backend.go
func New(stack *node.Node, config *ethconfig.Config) (*Ethereum, error) {
    ...
    if !config.TxPool.NoLocals {
        rejournal := config.TxPool.Rejournal
        if rejournal < time.Second {
            log.Warn("Sanitizing invalid txpool journal time", "provided", rejournal, "updated", time.Second)
            rejournal = time.Second
        }
        eth.localTxTracker = locals.New(config.TxPool.Journal, rejournal, eth.blockchain.Config(), eth.txPool)
        stack.RegisterLifecycle(eth.localTxTracker)
    }
    ...
}

// core/txpool/locals/tx_tracker.go

func (tracker *TxTracker) loop() {
	...
	for {
		select {
		case <-tracker.shutdownCh:
			return
		case <-timer.C:
			...
			if checkJournal {
				// Lock to prevent journal.rotate <-> journal.insert (via TrackAll) conflicts
				tracker.mu.Lock()
				lastJournal = time.Now()
				if err := tracker.journal.rotate(rejournal); err != nil {
					log.Warn("Transaction journal rotation failed", "err", err)
				}
				tracker.mu.Unlock()
			}
			timer.Reset(recheckInterval)
		}
	}
}

// core/txpool/legacypool/legacypool.go
// runReorg runs reset and promoteExecutables on behalf of scheduleReorgLoop.
func (pool *LegacyPool) runReorg(done chan struct{}, reset *txpoolResetRequest, dirtyAccounts *accountSet, events map[common.Address]*SortedMap) {
	...
	if len(events) > 0 {
		var txs []*types.Transaction
		for _, set := range events {
			txs = append(txs, set.Flatten()...)
		}
		pool.txFeed.Send(core.NewTxsEvent{Txs: txs})
	}
}

// eth/handler.go

func (h *handler) Start(maxPeers int) {
    ...
	h.txsCh = make(chan core.NewTxsEvent, txChanSize)
	h.txsSub = h.txpool.SubscribeTransactions(h.txsCh, false)
	go h.txBroadcastLoop()
    ...
}

// txBroadcastLoop announces new transactions to connected peers.
func (h *handler) txBroadcastLoop() {
	defer h.wg.Done()
	for {
		select {
		case event := <-h.txsCh:
			h.BroadcastTransactions(event.Txs)
		case <-h.txsSub.Err():
			return
		}
	}
}

//cmd/geth/main.go
func init() {
	// Initialize the CLI app and start Geth
	app.Action = geth
    ...
}

func main() {
	if err := app.Run(os.Args); err != nil {
		fmt.Fprintln(os.Stderr, err)
		os.Exit(1)
	}
}

// cmd/geth/main.go
func geth(ctx *cli.Context) error {
	if args := ctx.Args().Slice(); len(args) > 0 {
		return fmt.Errorf("invalid command: %q", args[0])
	}

	prepare(ctx)
	stack := makeFullNode(ctx)
	defer stack.Close()

	startNode(ctx, stack, false)
	stack.Wait()
	return nil
}

func startNode(ctx *cli.Context, stack *node.Node, isConsole bool) {
	// Start up the node itself
	utils.StartNode(ctx, stack, isConsole)
    ...
}
func StartNode(ctx *cli.Context, stack *node.Node, isConsole bool) {
	if err := stack.Start(); err != nil {
		Fatalf("Error starting protocol stack: %v", err)
	}
    ...
}

// node/node.go
func (n *Node) Start() error {
	...
	n.state = runningState
	// open networking and RPC endpoints
	err := n.openEndpoints()
	lifecycles := make([]Lifecycle, len(n.lifecycles))
	copy(lifecycles, n.lifecycles)
	n.lock.Unlock()
    ...
	var started []Lifecycle
	for _, lifecycle := range lifecycles {
		if err = lifecycle.Start(); err != nil {
			break
		}
		started = append(started, lifecycle)
	}
	...
	return err
}

// core/txpool/locals/tx_tracker.go
// Start implements node.Lifecycle interface
// Start is called after all services have been constructed and the networking
// layer was also initialized to spawn any goroutines required by the service.
func (tracker *TxTracker) Start() error {
	tracker.wg.Add(1)
	go tracker.loop()
	return nil
}

// core/txpool/locals/tx_tracker.go
func (tracker *TxTracker) loop() {
	defer tracker.wg.Done()

	if tracker.journal != nil {
		tracker.journal.load(func(transactions []*types.Transaction) []error {
			tracker.TrackAll(transactions)
			return nil
		})
		defer tracker.journal.close()
	}
	...
}     

// core/txpool/locals/journal.go
input, err := os.Open(journal.path)
if errors.Is(err, fs.ErrNotExist) { 
   return nil
}
if err != nil {
   return err
}
defer input.Close()

// core/txpool/locals/journal.go
journal.writer = new(devNull) 
defer func() { journal.writer = nil }()

// core/txpool/locals/journal.go
stream := rlp.NewStream(input, 0) 

// core/txpool/locals/journal.go
var (
   failure error
   batch   types.Transactions
)
loadBatch := func(txs types.Transactions) {
   for _, err := range add(txs) { 
      if err != nil {
         log.Debug("Failed to add journaled transaction", "err", err)
         dropped++
      }
   }
}

// core/txpool/locals/journal.go
for { 
   tx := new(types.Transaction) 
   if err = stream.Decode(tx); err != nil {
      if err != io.EOF {
         failure = err
      }
      if batch.Len() > 0 {
         loadBatch(batch)
      }
      break
   } 
   total++

   if batch = append(batch, tx); batch.Len() > 1024 {
      loadBatch(batch)
      batch = batch[:0]
   }
}

log.Info("Loaded local transaction journal", "transactions", total, "dropped", dropped)

// core/txpool/locals/journal.go
// 加载从磁盘解析交易日志转储，将其内容加载到指定的池中。
func (journal *journal) load(add func([]*types.Transaction) []error) error {
	// 打开 journal 文件，准备加载历史交易
	input, err := os.Open(journal.path)
	if errors.Is(err, fs.ErrNotExist) {
		// 如果 journal 文件根本不存在，则跳过解析，直接返回
		return nil
	}
	if err != nil {
		// 其他打开错误则直接返回
		return err
	}
	defer input.Close() // 确保函数结束时关闭文件

	// 临时禁用 journal 写入（防止加载时重复写入）
	journal.writer = new(devNull)
	defer func() { journal.writer = nil }() // 函数退出前恢复 writer

	// 将 journal 中的所有交易注入交易池
	stream := rlp.NewStream(input, 0) // 创建 RLP 解码流
	total, dropped := 0, 0            // 统计总交易数和被丢弃的交易数

	// 定义一个内部函数：用于加载一小批交易，并更新计数器
	loadBatch := func(txs types.Transactions) {
		for _, err := range add(txs) { // 调用外部传入的 add 函数添加交易
			if err != nil {
				log.Debug("Failed to add journaled transaction", "err", err)
				dropped++ // 记录失败数量
			}
		}
	}

	var (
		failure error                    // 记录最终的解析错误（非 EOF）
		batch   types.Transactions       // 当前批次的交易缓存
	)

	// 循环读取并解析每笔交易
	for {
		// 解析下一笔交易，出错则终止循环
		tx := new(types.Transaction)
		if err = stream.Decode(tx); err != nil {
			if err != io.EOF { // 如果不是正常结束（EOF），记录错误
				failure = err
			}
			// 如果当前批次还有未处理交易，先处理完
			if batch.Len() > 0 {
				loadBatch(batch)
			}
			break // 退出循环
		}

		// 成功解析一笔交易，加入批次
		total++ // 总交易数 +1

		batch = append(batch, tx)
		// 如果批次大小超过 1024，则立即处理并清空批次
		if batch.Len() > 1024 {
			loadBatch(batch)
			batch = batch[:0] // 重置批次，保留底层数组以复用内存
		}
	}

	// 加载完成后打印日志：共加载多少笔，丢弃多少笔
	log.Info("Loaded local transaction journal", "transactions", total, "dropped", dropped)

	return failure // 返回最终错误（如解析格式错误等）
}

// core/txpool/locals/journal.go
// insert将指定事务添加到本地磁盘日志中。
func (journal *journal) insert(tx *types.Transaction) error {
	if journal.writer == nil {
		return errNoActiveJournal
	}
	if err := rlp.Encode(journal.writer, tx); err != nil {
		return err
	}
	return nil
}

func (tracker *TxTracker) loop() {
	...
	var (
		lastJournal = time.Now()
		timer       = time.NewTimer(10 * time.Second) // Do initial check after 10 seconds, do rechecks more seldom.
	)
	for {
		select {
		case <-tracker.shutdownCh:
			return
		case <-timer.C:
			checkJournal := tracker.journal != nil && time.Since(lastJournal) > tracker.rejournal
			resubmits, rejournal := tracker.recheck(checkJournal)
			if len(resubmits) > 0 {
				tracker.pool.Add(resubmits, false)
			}
			if checkJournal {
				// Lock to prevent journal.rotate <-> journal.insert (via TrackAll) conflicts
				tracker.mu.Lock()
				lastJournal = time.Now()
				if err := tracker.journal.rotate(rejournal); err != nil {
					log.Warn("Transaction journal rotation failed", "err", err)
				}
				tracker.mu.Unlock()
			}
			...
		}
	}
}

// core/txpool/locals/journal.go
// Close the current journal (if any is open)
if journal.writer != nil {
	if err := journal.writer.Close(); err != nil {
		return err
	}
	journal.writer = nil
}
// Generate a new journal with the contents of the current pool
replacement, err := os.OpenFile(journal.path+".new", os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
if err != nil {
	return err
}
journaled := 0
for _, txs := range all {
	for _, tx := range txs {
		if err = rlp.Encode(replacement, tx); err != nil {
			replacement.Close()
			return err
		}
	}
	journaled += len(txs)
}
replacement.Close()

// Replace the live journal with the newly generated one
if err = os.Rename(journal.path+".new", journal.path); err != nil {
	return err
}
sink, err := os.OpenFile(journal.path, os.O_WRONLY|os.O_APPEND, 0644)
if err != nil {
	return err
}
journal.writer = sink
logger := log.Info
if len(all) == 0 {
	logger = log.Debug
}
logger("Regenerated local transaction journal", "transactions", journaled, "accounts", len(all))

// rotate regenerates the transaction journal based on the current contents of
// the transaction pool.
func (journal *journal) rotate(all map[common.Address]types.Transactions) error {
	// Close the current journal (if any is open)
	if journal.writer != nil {
		if err := journal.writer.Close(); err != nil {
			return err
		}
		journal.writer = nil
	}
	// Generate a new journal with the contents of the current pool
	replacement, err := os.OpenFile(journal.path+".new", os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
	if err != nil {
		return err
	}
	journaled := 0
	for _, txs := range all {
		for _, tx := range txs {
			if err = rlp.Encode(replacement, tx); err != nil {
				replacement.Close()
				return err
			}
		}
		journaled += len(txs)
	}
	replacement.Close()

	// Replace the live journal with the newly generated one
	if err = os.Rename(journal.path+".new", journal.path); err != nil {
		return err
	}
	sink, err := os.OpenFile(journal.path, os.O_WRONLY|os.O_APPEND, 0644)
	if err != nil {
		return err
	}
	journal.writer = sink

	logger := log.Info
	if len(all) == 0 {
		logger = log.Debug
	}
	logger("Regenerated local transaction journal", "transactions", journaled, "accounts", len(all))

	return nil
}

func (tracker *TxTracker) loop() {
	...
	var (
		lastJournal = time.Now()
		timer       = time.NewTimer(10 * time.Second) // Do initial check after 10 seconds, do rechecks more seldom.
	)
	for {
		select {
		case <-tracker.shutdownCh:
			return
		case <-timer.C:
			checkJournal := tracker.journal != nil && time.Since(lastJournal) > tracker.rejournal
			resubmits, rejournal := tracker.recheck(checkJournal)
			if len(resubmits) > 0 {
				tracker.pool.Add(resubmits, false)
			}
			if checkJournal {
				// Lock to prevent journal.rotate <-> journal.insert (via TrackAll) conflicts
				tracker.mu.Lock()
				lastJournal = time.Now()
				if err := tracker.journal.rotate(rejournal); err != nil {
					log.Warn("Transaction journal rotation failed", "err", err)
				}
				tracker.mu.Unlock()
			}
			timer.Reset(recheckInterval)
		}
	}
}

func (tracker *TxTracker) loop() {
	// ...
	for {
		select {
		// ...
		case <-timer.C:
			// ...
			if len(resubmits) > 0 {
				tracker.pool.Add(resubmits, false)
			}
			// ...
		}
	}
}

func (p *TxPool) Add(txs []*types.Transaction, sync bool) []error {
	// ...
    // txsets 会根据交易类型被分到不同的集合
	for i := 0; i < len(p.subpools); i++ {
		errsets[i] = p.subpools[i].Add(txsets[i], sync)
	}
	// ...
	return errs
}

// Add enqueues a batch of transactions into the pool if they are valid.
func (pool *LegacyPool) Add(txs []*types.Transaction, sync bool) []error {
	var (
		errs = make([]error, len(txs))
		news = make([]*types.Transaction, 0, len(txs)) // 收集通过初筛的新交易
	)
	for i, tx := range txs {
		if pool.all.Get(tx.Hash()) != nil {
			errs[i] = txpool.ErrAlreadyKnown
			continue
		}
		if err := pool.ValidateTxBasics(tx); err != nil {
			errs[i] = err
			continue
		}
		news = append(news, tx)
	}
	if len(news) == 0 {
		return errs
	}

	// 对所有新交易进行加锁和核心处理
	pool.mu.Lock()
	newErrs, dirtyAddrs := pool.addTxsLocked(news)
	pool.mu.Unlock()

    // 合并错误并触发后续处理
    // ...
	done := pool.requestPromoteExecutables(dirtyAddrs) // 触发可执行交易的提升检查
	// ...
	return errs
}

// ValidateTxBasics checks whether a transaction is valid according to the consensus rules.
func (pool *LegacyPool) ValidateTxBasics(tx *types.Transaction) error {
	opts := &txpool.ValidationOptions{
		Config: pool.chainconfig,
		Accept: 0 | // 定义节点接受的交易类型位掩码
			1<<types.LegacyTxType |
			1<<types.AccessListTxType |
			1<<types.DynamicFeeTxType |
			1<<types.SetCodeTxType,
		MaxSize: txMaxSize,
		MinTip:  pool.gasTip.Load().ToBig(), // 检查是否满足节点的最低 Gas Tip 要求
	}
	return txpool.ValidateTransaction(tx, pool.currentHead.Load(), pool.signer, opts)
}

// addTxsLocked attempts to queue a batch of transactions if they are valid.
func (pool *LegacyPool) addTxsLocked(txs []*types.Transaction) ([]error, *accountSet) {
	dirty := newAccountSet(pool.signer) // 记录受影响的账户，用于后续处理
	errs := make([]error, len(txs))
	for i, tx := range txs {
		replaced, err := pool.add(tx)
		errs[i] = err
		if err == nil && !replaced {
			dirty.addTx(tx) // 记录成功添加新交易的账户
		}
	}
	return errs, dirty
}

// core/txpool/legacypool.go
if !hasPending && !hasQueued {
    if err := pool.reserver.Hold(from); err != nil { // ❶ 账户预留机制
        return false, err
    }
}

// 简化后的容量管理逻辑
if uint64(pool.all.Slots()+numSlots(tx)) > pool.config.GlobalSlots+pool.config.GlobalQueue { // ❷ 容量检查
    if pool.priced.Underpriced(tx) { // ❸ 价格门槛
        return false, txpool.ErrUnderpriced
    }
    if pool.changesSinceReorg > int(pool.config.GlobalSlots/4) { // ❹ 防抖动机制
        return false, ErrTxPoolOverflow
    }
    drop, success := pool.priced.Discard(...) // ❺ 准备驱逐
  
    if pool.isGapped(from, tx) { // ❻ 驱逐保护
        if replacesPending { // 如果这个未来交易会导致一个 pending 交易被驱逐
            return false, ErrFutureReplacePending // 则拒绝它
        }
    }
    // ... 执行驱逐 ...
    for _, tx := range drop {
        pool.removeTx(tx.Hash(), false, ...)
    }
}

    if list := pool.pending[from]; list != nil && list.Contains(tx.Nonce()) { // ❼
        inserted, old := list.Add(tx, pool.config.PriceBump) // ❽
        if !inserted {
            return false, txpool.ErrReplaceUnderpriced
        }
        if old != nil { // ❾ 替换成功，移除旧交易
            pool.all.Remove(old.Hash())
            pool.priced.Removed(1)
        }
        pool.all.Add(tx)
        pool.priced.Put(tx)
        return old != nil, nil
    }