{"content":{"title":"optimism sequencer背后的魔法（四）：op-batcher工作原理","body":"原文链接：https://github.com/joohhnnn/Understanding-Optimism-Codebase-CN/blob/main/sequencer/03-how-batcher-works.md\r\n作者：[joohhnnn](https://github.com/joohhnnn)\r\n\r\n# batcher工作原理\r\n在这一章节中，我们将探讨到底什么是`batcher` ⚙️\r\n官方specs中有batcher的介绍([source](https://github.com/ethereum-optimism/optimism/blob/develop/specs/batcher.md))\r\n\r\n在进行之前，我们先提出几个问题，通过这两个问题来真正理解`batcher`的作用以及工作原理\r\n- `batcher`是什么？它为什么叫做`batcher`\r\n- `batcher`在代码中到底是怎么运行的？\r\n\r\n## 前置知识\r\n- 在rollup机制中，要想做到的去中心化特性，例如抗审查等。我们必须要把layer2上发生的数据（transactions）全部发送到layer1当中。这样就可以在利用layer1的安全性的同时，又可以完全从layer1中构建出来整个layer2的数据，使得layer2才真正的具有有效性。\r\n- [Epochs and the Sequencing Window](https://github.com/ethereum-optimism/optimism/blob/develop/specs/overview.md#epochs-and-the-sequencing-window):`Epoch`可以简单理解为L1新的一个`区块（N+1）`生成的这段时间。`epoch`的编号等于L1`区块N`的编号，在L1区块`N -> N+1` 这段时间内产生的所有L2区块都属于`epoch N`。在上个概念中我们提到必须上传L2的数据到L1中，那么我们应该在什么范围内上传数据才是有效的呢，`Sequencing Window`的size给了我们答案，即区块N/epoch N的相关数据，必须在L1的第`N + size`之前已经上传到L1了。\r\n- Batch/Batcher Transaction: `Batch`可以简单理解为每一个L2区块构建所需要的交易。`Batcher Transaction`为多个batch组合起来经过加工后发送到L1的那笔交易\r\n- [Channe](https://github.com/ethereum-optimism/optimism/blob/develop/specs/glossary.md#channel): `channel`可以简单理解为是`batch`的组合，组合是为了获得更好的压缩率，从而降低数据可用性成本，以使`batcher`上传的成本进一步降低。\r\n- [Frame](https://github.com/ethereum-optimism/optimism/blob/develop/specs/glossary.md#channel-frame): `frame`可以理解为，有时候为了更好的压缩率，可能会导致`channel`数据过大而不能直接被`batcher`将整个`channel`发送给L1，因此需要对`channel`进行切割，分多次进行发送。\r\n\r\n## 什么是batcher\r\n在rollup中，需要一个角色来传递L2信息到L1当中，同时每当有新的交易就马上发送是昂贵且不方便管理的。这时候我们将需要制定一种合理的批量上传策略。因此，为了解决这个问题，batcher出现了。batcher是唯一存在（sequencer当前掌管私钥），且和特定地址发送`Batcher Transaction`来传递L2信息的组件。\r\n\r\nbatcher通过对unsafe区块数据进行收集，来获取多个batch，在这里每个区块都对应一个batch。当收集足够的batch进行高效压缩后生成channel，并以frame的形式发送到L1来完成L2的信息上传。\r\n\r\n## 代码实现\r\n在这部分我们会从代码层来进行深度的机制和实现原理的讲解\r\n\r\n### 程序起点 \r\n`op-batcher/batcher/driver.go`\r\n\r\n通过调用`Start`函数来启动`loop`循环，在`loop`的循环中，主要处理三件事\r\n- 当定时器触发时，将所有新的还未加载的`L2block`加载进来，然后触发`publishStateToL1`函数向L1进行`state`发布\r\n- 处理`receipts`，记录成功或者失败状态\r\n- 处理关闭请求\r\n\r\n```go\r\n    func (l *BatchSubmitter) Start() error {\r\n        l.log.Info(\"Starting Batch Submitter\")\r\n\r\n        l.mutex.Lock()\r\n        defer l.mutex.Unlock()\r\n\r\n        if l.running {\r\n            return errors.New(\"batcher is already running\")\r\n        }\r\n        l.running = true\r\n\r\n        l.shutdownCtx, l.cancelShutdownCtx = context.WithCancel(context.Background())\r\n        l.killCtx, l.cancelKillCtx = context.WithCancel(context.Background())\r\n        l.state.Clear()\r\n        l.lastStoredBlock = eth.BlockID{}\r\n\r\n        l.wg.Add(1)\r\n        go l.loop()\r\n\r\n        l.log.Info(\"Batch Submitter started\")\r\n\r\n        return nil\r\n    }\r\n```\r\n\r\n```go\r\n    func (l *BatchSubmitter) loop() {\r\n        defer l.wg.Done()\r\n\r\n        ticker := time.NewTicker(l.PollInterval)\r\n        defer ticker.Stop()\r\n\r\n        receiptsCh := make(chan txmgr.TxReceipt[txData])\r\n        queue := txmgr.NewQueue[txData](l.killCtx, l.txMgr, l.MaxPendingTransactions)\r\n\r\n        for {\r\n            select {\r\n            case <-ticker.C:\r\n                if err := l.loadBlocksIntoState(l.shutdownCtx); errors.Is(err, ErrReorg) {\r\n                    err := l.state.Close()\r\n                    if err != nil {\r\n                        l.log.Error(\"error closing the channel manager to handle a L2 reorg\", \"err\", err)\r\n                    }\r\n                    l.publishStateToL1(queue, receiptsCh, true)\r\n                    l.state.Clear()\r\n                    continue\r\n                }\r\n                l.publishStateToL1(queue, receiptsCh, false)\r\n            case r := <-receiptsCh:\r\n                l.handleReceipt(r)\r\n            case <-l.shutdownCtx.Done():\r\n                err := l.state.Close()\r\n                if err != nil {\r\n                    l.log.Error(\"error closing the channel manager\", \"err\", err)\r\n                }\r\n                l.publishStateToL1(queue, receiptsCh, true)\r\n                return\r\n            }\r\n        }\r\n    }\r\n```\r\n\r\n### 加载最新区块数据\r\n`op-batcher/batcher/driver.go`\r\n\r\n`loadBlocksIntoState`函数调用`calculateL2BlockRangeToStore`来获取自上次发送`batch transaction`而派生的最新`safeblock`后新生成的`unsafeblock`范围。然后循环将这个范围中的每一个`unsafe`块调用`loadBlockIntoState`函数从L2里获取并通过`AddL2Block`函数加载到内部的`block队列`里。等待进一步处理。\r\n\r\n```go\r\n    func (l *BatchSubmitter) loadBlocksIntoState(ctx context.Context) error {\r\n        start, end, err := l.calculateL2BlockRangeToStore(ctx)\r\n        ……\r\n        var latestBlock *types.Block\r\n        // Add all blocks to \"state\"\r\n        for i := start.Number + 1; i < end.Number+1; i++ {\r\n            block, err := l.loadBlockIntoState(ctx, i)\r\n            if errors.Is(err, ErrReorg) {\r\n                l.log.Warn(\"Found L2 reorg\", \"block_number\", i)\r\n                l.lastStoredBlock = eth.BlockID{}\r\n                return err\r\n            } else if err != nil {\r\n                l.log.Warn(\"failed to load block into state\", \"err\", err)\r\n                return err\r\n            }\r\n            l.lastStoredBlock = eth.ToBlockID(block)\r\n            latestBlock = block\r\n        }\r\n        ……\r\n    }\r\n```\r\n\r\n```go\r\n    func (l *BatchSubmitter) loadBlockIntoState(ctx context.Context, blockNumber uint64) (*types.Block, error) {\r\n        ……\r\n        block, err := l.L2Client.BlockByNumber(ctx, new(big.Int).SetUint64(blockNumber))\r\n        ……\r\n        if err := l.state.AddL2Block(block); err != nil {\r\n            return nil, fmt.Errorf(\"adding L2 block to state: %w\", err)\r\n        }\r\n        ……\r\n        return block, nil\r\n    }\r\n```\r\n### 将加载的block数据处理，并发送到layer1\r\n`op-batcher/batcher/driver.go`\r\n\r\n`publishTxToL1`函数使用`TxData`函数对之前加载到数据进行处理，并调用`sendTransaction`函数发送到L1\r\n\r\n```go\r\n    func (l *BatchSubmitter) publishTxToL1(ctx context.Context, queue *txmgr.Queue[txData], receiptsCh chan txmgr.TxReceipt[txData]) error {\r\n        // send all available transactions\r\n        l1tip, err := l.l1Tip(ctx)\r\n        if err != nil {\r\n            l.log.Error(\"Failed to query L1 tip\", \"error\", err)\r\n            return err\r\n        }\r\n        l.recordL1Tip(l1tip)\r\n\r\n        // Collect next transaction data\r\n        txdata, err := l.state.TxData(l1tip.ID())\r\n        if err == io.EOF {\r\n            l.log.Trace(\"no transaction data available\")\r\n            return err\r\n        } else if err != nil {\r\n            l.log.Error(\"unable to get tx data\", \"err\", err)\r\n            return err\r\n        }\r\n\r\n        l.sendTransaction(txdata, queue, receiptsCh)\r\n        return nil\r\n    }\r\n```\r\n\r\n#### TxData详解\r\n`op-batcher/batcher/channel_manager.go`\r\n\r\n`TxData`函数主要负责两件事务\r\n- 查找第一个含有`frame`的的`channel`，如果存在且通过检查后使用`nextTxData`获取数据并返回\r\n- 如果没有这样的`channel`，我们需要现调用`ensureChannelWithSpace`检查`channel`还有剩余的空间，再使用`processBlocks`将之前加载到`block队列`中的数据构造到 `outchannel的composer`当中压缩\r\n- `outputFrames`将`outchannel composer`当中的数据切割成适合大小的`frame`\r\n- 最后再把刚构造到数据通过`nextTxData`函数返回出去。\r\n\r\n\r\n`EnsureChannelWithSpace` 确保 `currentChannel` 填充有可容纳更多数据的空间的`channel`（即，`channel.IsFull` 返回 `false`）。 如果 `currentChannel` 为零或已满，则会创建一个新`channel`。\r\n\r\n```go\r\n    func (s *channelManager) TxData(l1Head eth.BlockID) (txData, error) {\r\n        s.mu.Lock()\r\n        defer s.mu.Unlock()\r\n        var firstWithFrame *channel\r\n        for _, ch := range s.channelQueue {\r\n            if ch.HasFrame() {\r\n                firstWithFrame = ch\r\n                break\r\n            }\r\n        }\r\n\r\n        dataPending := firstWithFrame != nil && firstWithFrame.HasFrame()\r\n        s.log.Debug(\"Requested tx data\", \"l1Head\", l1Head, \"data_pending\", dataPending, \"blocks_pending\", len(s.blocks))\r\n\r\n        // Short circuit if there is a pending frame or the channel manager is closed.\r\n        if dataPending || s.closed {\r\n            return s.nextTxData(firstWithFrame)\r\n        }\r\n\r\n        // No pending frame, so we have to add new blocks to the channel\r\n\r\n        // If we have no saved blocks, we will not be able to create valid frames\r\n        if len(s.blocks) == 0 {\r\n            return txData{}, io.EOF\r\n        }\r\n\r\n        if err := s.ensureChannelWithSpace(l1Head); err != nil {\r\n            return txData{}, err\r\n        }\r\n\r\n        if err := s.processBlocks(); err != nil {\r\n            return txData{}, err\r\n        }\r\n\r\n        // Register current L1 head only after all pending blocks have been\r\n        // processed. Even if a timeout will be triggered now, it is better to have\r\n        // all pending blocks be included in this channel for submission.\r\n        s.registerL1Block(l1Head)\r\n\r\n        if err := s.outputFrames(); err != nil {\r\n            return txData{}, err\r\n        }\r\n\r\n        return s.nextTxData(s.currentChannel)\r\n    }\r\n\r\n```\r\n\r\n`processBlocks`函数在内部通过`AddBlock`把`block队列`里的`block`加入到当前的`channel`当中\r\n\r\n```go\r\n    func (s *channelManager) processBlocks() error {\r\n        var (\r\n            blocksAdded int\r\n            _chFullErr  *ChannelFullError // throw away, just for type checking\r\n            latestL2ref eth.L2BlockRef\r\n        )\r\n        for i, block := range s.blocks {\r\n            l1info, err := s.currentChannel.AddBlock(block)\r\n            if errors.As(err, &_chFullErr) {\r\n                // current block didn't get added because channel is already full\r\n                break\r\n            } else if err != nil {\r\n                return fmt.Errorf(\"adding block[%d] to channel builder: %w\", i, err)\r\n            }\r\n            s.log.Debug(\"Added block to channel\", \"channel\", s.currentChannel.ID(), \"block\", block)\r\n\r\n            blocksAdded += 1\r\n            latestL2ref = l2BlockRefFromBlockAndL1Info(block, l1info)\r\n            s.metr.RecordL2BlockInChannel(block)\r\n            // current block got added but channel is now full\r\n            if s.currentChannel.IsFull() {\r\n                break\r\n            }\r\n        }\r\n```\r\n\r\n`AddBlock` 首先通过`BlockToBatch`把`batch`从`blcok`中获取出来，再通过`AddBatch`函数对数据进行压缩并存储。\r\n\r\n```go\r\n    func (c *channelBuilder) AddBlock(block *types.Block) (derive.L1BlockInfo, error) {\r\n        if c.IsFull() {\r\n            return derive.L1BlockInfo{}, c.FullErr()\r\n        }\r\n\r\n        batch, l1info, err := derive.BlockToBatch(block)\r\n        if err != nil {\r\n            return l1info, fmt.Errorf(\"converting block to batch: %w\", err)\r\n        }\r\n\r\n        if _, err = c.co.AddBatch(batch); errors.Is(err, derive.ErrTooManyRLPBytes) || errors.Is(err, derive.CompressorFullErr) {\r\n            c.setFullErr(err)\r\n            return l1info, c.FullErr()\r\n        } else if err != nil {\r\n            return l1info, fmt.Errorf(\"adding block to channel out: %w\", err)\r\n        }\r\n        c.blocks = append(c.blocks, block)\r\n        c.updateSwTimeout(batch)\r\n\r\n        if err = c.co.FullErr(); err != nil {\r\n            c.setFullErr(err)\r\n            // Adding this block still worked, so don't return error, just mark as full\r\n        }\r\n\r\n        return l1info, nil\r\n    }\r\n```\r\n\r\n在`txdata`获取后，使用`sendTransaction`将整个数据发送到L1当中。\r\n\r\n## 总结\r\n在这一章节中，我们了解了什么是`batcher`并且了解了`batcher`的运行原理，你可以在这个 [address](https://etherscan.io/address/0x6887246668a3b87f54deb3b94ba47a6f63f32985)中查看当前`batcher`的行为。\r\n\r\n---\r\n[第一章](https://learnblockchain.cn/article/6589) | [第二章](https://learnblockchain.cn/article/6755) | [第三章](https://learnblockchain.cn/article/6756) | [第四章](https://learnblockchain.cn/article/6757) | [第五章](https://learnblockchain.cn/article/6758) |"},"author":{"user":"https://learnblockchain.cn/people/4858","address":null},"history":"QmZg8G7Wp6fyopRcapZo9Ca7Y6PmhdBE2V5bNyYHK3oXs2","timestamp":1698110931,"version":1}