{"content":{"title":"optimism sequencer背后的魔法（二）：op-stack中区块的传递","body":"原文链接：https://github.com/joohhnnn/Understanding-Optimism-Codebase-CN/blob/main/sequencer/01-how-block-sync.md\r\n作者：[joohhnnn](https://github.com/joohhnnn)\r\n# optimism中区块的传递\r\n\r\n区块的传递是整个optimism rollup系统中较为重要的概念，在这一章节，我们将从介绍optimism中多种sync方式的原理，来揭开整个系统里区块的传递过程。\r\n\r\n## 区块类型\r\n\r\n在进行进一步深入前，让我们了解一些基本的概念。\r\n\r\n- **Unsafe L2 Block (不安全的 L2 区块)**:\r\n    - 这是指 L1 链上最高的 L2 区块，其 L1 起源是规范 L1 链的 *可能* 扩展（如 op-node 所知）。这意味着，尽管该区块链接到 L1 链，但其完整性和正确性尚未得到充分验证。\r\n\r\n- **Safe L2 Block (安全的 L2 区块)**:\r\n    - 这是指 L1 链上最高的 L2 区块，其 epoch 的序列窗口在规范的 L1 链中是完整的（如 op-node 所知）。这意味着该区块的所有前提条件都已在 L1 链上得到验证，因此它被认为是安全的。\r\n\r\n- **Finalized L2 Block (定稿的 L2 区块)**:\r\n    - 这是指已知完全源自定稿 L1 区块数据的 L2 区块。这意味着该区块不仅安全，而且已根据 L1 链的数据完全确认，不会再发生更改。\r\n\r\n\r\n## sync类型\r\n\r\n1. **op-node p2p gossip 同步**：\r\n   - op-node 通过 p2p gossip 协议接收最新的不安全区块，由 sequencer 推送的。\r\n\r\n2. **op-node 基于libp2p的请求-响应的逆向区块头同步**：\r\n   - 通过此同步方式，op-node 可以填补不安全区块的任何缺口。\r\n\r\n3. **执行层（EL，又名 engine sync）同步**：\r\n   - 在 op-node 中有两个标志，允许来自 gossip 的不安全区块触发引擎中向这些区块的长范围同步。相关的标志是 `--l2.engine-sync` 和 `--l2.skip-sync-start-check`（用于处理非常旧的安全区块）。然后，如果为此设置了 EL，它可以执行任何同步，例如 snap-sync（需要 op-geth p2p 连接等，并且需要从某些节点进行同步）。\r\n\r\n4. **op-node RPC 同步**：\r\n   - 这是一种基于可信 RPC 方法的同步，当 L1 出现问题时，这种同步方式相对简单。\r\n\r\n## op-node p2p gossip 同步\r\n\r\n这种同步的场景处于：当l2的块新产生的时候，即在上一节我们讨论的sequencer模式下是如何产生新的区块的。\r\n\r\n当产生新的区块后，sequencer通过基于libp2p的P2P网络的pub/sub（广播/订阅）模块，向’新unsafe区块‘ topic 发出广播。所有订阅了此topic的节点都会直接或间接的收到这一广播消息。[详情可以查看](https://github.com/joohhnnn/Understanding-Optimism-Codebase-CN/blob/main/sequencer/02-how-optimism-use-libp2p.md#gossip%E4%B8%8B%E7%9A%84%E5%8C%BA%E5%9D%97%E4%BC%A0%E6%92%AD)\r\n\r\n## op-node 基于libp2p的请求-响应的逆向区块头同步\r\n\r\n这种同步的场景处于：当节点因为特殊情况，比如宕机后重新链接，可能会产生一些没有同步上的区块（gaps）\r\n\r\n当这种情况出现的时候，可以通过p2p网络的反向链的方式快速同步，即通过使用libp2p原生的stream流来和其他p2p节点建立链接，同时发送同步请求。[详情可以查看](https://github.com/joohhnnn/Understanding-Optimism-Codebase-CN/blob/main/sequencer/02-how-optimism-use-libp2p.md#%E5%BD%93%E5%AD%98%E5%9C%A8%E7%BC%BA%E5%A4%B1%E5%8C%BA%E5%9D%97%E9%80%9A%E8%BF%87p2p%E5%BF%AB%E9%80%9F%E5%90%8C%E6%AD%A5)\r\n\r\n## 执行层（EL，又名 engine sync）同步\r\n\r\n这种同步的场景处于：当有较多区块，一个大范围区块需要同步的时候，从l1慢慢派生比较慢，想要快速同步。\r\n\r\n使用`--l2.engine-sync` 和 `--l2.skip-sync-start-check`去启动op-node，发送的payload来达到发送长范围同步请求的目的。\r\n\r\n### 代码层讲解\r\n\r\n首先我们来看一下这两个标志的定义\r\n\r\n在 `op-node/flags/flags.go` 中定义并解释了这两个flag的作用\r\n\r\n- **L2EngineSyncEnabled Flag (`l2.engine-sync`)**:\r\n    - 该标志用于启用或禁用执行引擎的 P2P 同步功能。当设置为 `true` 时，它允许执行引擎通过 P2P 网络与其他节点同步区块数据。它的默认值是 `false`，意味着在默认情况下，该 P2P 同步功能是禁用的。\r\n\r\n- **SkipSyncStartCheck Flag (`l2.skip-sync-start-check`)**:\r\n    - 该标志用于在确定同步起始点时，跳过对不安全 L2 区块的 L1 起源一致性的合理性检查。当设置为 `true` 时，它会推迟 L1 起源的验证。如果你正在使用 `l2.engine-sync`，建议启用此标志来跳过初始的一致性检查。它的默认值是 `false`，意味着在默认情况下，该合理性检查是启用的。\r\n\r\n```go\r\n\tL2EngineSyncEnabled = &cli.BoolFlag{\r\n\t\tName:     \"l2.engine-sync\",\r\n\t\tUsage:    \"Enables or disables execution engine P2P sync\",\r\n\t\tEnvVars:  prefixEnvVars(\"L2_ENGINE_SYNC_ENABLED\"),\r\n\t\tRequired: false,\r\n\t\tValue:    false,\r\n\t}\r\n\tSkipSyncStartCheck = &cli.BoolFlag{\r\n\t\tName: \"l2.skip-sync-start-check\",\r\n\t\tUsage: \"Skip sanity check of consistency of L1 origins of the unsafe L2 blocks when determining the sync-starting point. \" +\r\n\t\t\t\"This defers the L1-origin verification, and is recommended to use in when utilizing l2.engine-sync\",\r\n\t\tEnvVars:  prefixEnvVars(\"L2_SKIP_SYNC_START_CHECK\"),\r\n\t\tRequired: false,\r\n\t\tValue:    false,\r\n\t}\r\n```\r\n#### L2EngineSyncEnabled\r\n\r\n`L2EngineSyncEnabled`标志用于在op-node接收到新的`unsafe`的payload（区块）后，发送给op-geth进一步验证时，触发op-geth的p2p之间sync，在sync期间所有的`unsafe`区块都会被视为通过验证，并进行下一个unsafe的流程。op-geth内部的p2p sync比较适用于长范围的`unsafe`区块的获取。其实在op-geth内部，不管`L2EngineSyncEnabled`标志有没有启用，在遇到parent区块不存在的时候，都会开启sync去同步数据。\r\n\r\n让我们深入代码层面看一下\r\n首先是 `op-node/rollup/derive/engine_queue.go`\r\n\r\n`EngineSync`为`L2EngineSyncEnabled`标志的具体表达。在这里嵌套在两个检查函数当中。\r\n\r\n```go\r\n   // checkNewPayloadStatus checks returned status of engine_newPayloadV1 request for next unsafe payload.\r\n   // It returns true if the status is acceptable.\r\n   func (eq *EngineQueue) checkNewPayloadStatus(status eth.ExecutePayloadStatus) bool {\r\n      if eq.syncCfg.EngineSync {\r\n         // Allow SYNCING and ACCEPTED if engine P2P sync is enabled\r\n         return status == eth.ExecutionValid || status == eth.ExecutionSyncing || status == eth.ExecutionAccepted\r\n      }\r\n      return status == eth.ExecutionValid\r\n   }\r\n\r\n   // checkForkchoiceUpdatedStatus checks returned status of engine_forkchoiceUpdatedV1 request for next unsafe payload.\r\n   // It returns true if the status is acceptable.\r\n   func (eq *EngineQueue) checkForkchoiceUpdatedStatus(status eth.ExecutePayloadStatus) bool {\r\n      if eq.syncCfg.EngineSync {\r\n         // Allow SYNCING if engine P2P sync is enabled\r\n         return status == eth.ExecutionValid || status == eth.ExecutionSyncing\r\n      }\r\n      return status == eth.ExecutionValid\r\n   }\r\n```\r\n\r\n让我们把视角转到op-geth的 `eth/catalyst/api.go`当中，当parent区块缺失后，触发sync，并且返回`SYNCING Status`\r\n\r\n```go\r\n   func (api *ConsensusAPI) newPayload(params engine.ExecutableData) (engine.PayloadStatusV1, error) {\r\n      …\r\n      // If the parent is missing, we - in theory - could trigger a sync, but that\r\n      // would also entail a reorg. That is problematic if multiple sibling blocks\r\n      // are being fed to us, and even more so, if some semi-distant uncle shortens\r\n      // our live chain. As such, payload execution will not permit reorgs and thus\r\n      // will not trigger a sync cycle. That is fine though, if we get a fork choice\r\n      // update after legit payload executions.\r\n      parent := api.eth.BlockChain().GetBlock(block.ParentHash(), block.NumberU64()-1)\r\n      if parent == nil {\r\n         return api.delayPayloadImport(block)\r\n      }\r\n      …\r\n   }\r\n```\r\n\r\n```go\r\n   func (api *ConsensusAPI) delayPayloadImport(block *types.Block) (engine.PayloadStatusV1, error) {\r\n      …\r\n      if err := api.eth.Downloader().BeaconExtend(api.eth.SyncMode(), block.Header()); err == nil {\r\n         log.Debug(\"Payload accepted for sync extension\", \"number\", block.NumberU64(), \"hash\", block.Hash())\r\n         return engine.PayloadStatusV1{Status: engine.SYNCING}, nil\r\n      }\r\n      …\r\n   }\r\n```\r\n#### SkipSyncStartCheck \r\n`SkipSyncStartCheck`这个标识符主要是帮助在选择sync模式下，优化性能和减少不必要的检查。在已确认找到一个符合条件的L2块后，代码会跳过进一步的健全性检查，以加速同步或其他后续处理。这是一种优化手段，用于在确定性高的情况下快速地进行操作。\r\n\r\n在`op-node/rollup/sync/start.go`目录中\r\n\r\n`FindL2Heads`函数通过从给定的“开始”（start）点（即之前的不安全L2区块）开始逐步回溯，来查找这三种类型的区块。在回溯过程中，该函数会检查各个L2区块的L1源是否与已知的L1规范链匹配，以及是否符合其他一些条件和检查。这允许函数更快地确定L2的“安全”头部，从而可能加速整个同步过程。\r\n\r\n```go\r\n   func FindL2Heads(ctx context.Context, cfg *rollup.Config, l1 L1Chain, l2 L2Chain, lgr log.Logger, syncCfg *Config) (result *FindHeadsResult, err error) {\r\n      …\r\n      for {\r\n\r\n         …\r\n\r\n         if syncCfg.SkipSyncStartCheck && highestL2WithCanonicalL1Origin.Hash == n.Hash {\r\n            lgr.Info(\"Found highest L2 block with canonical L1 origin. Skip further sanity check and jump to the safe head\")\r\n            n = result.Safe\r\n            continue\r\n         }\r\n         // Pull L2 parent for next iteration\r\n         parent, err := l2.L2BlockRefByHash(ctx, n.ParentHash)\r\n         if err != nil {\r\n            return nil, fmt.Errorf(\"failed to fetch L2 block by hash %v: %w\", n.ParentHash, err)\r\n         }\r\n\r\n         // Check the L1 origin relation\r\n         if parent.L1Origin != n.L1Origin {\r\n            // sanity check that the L1 origin block number is coherent\r\n            if parent.L1Origin.Number+1 != n.L1Origin.Number {\r\n               return nil, fmt.Errorf(\"l2 parent %s of %s has L1 origin %s that is not before %s\", parent, n, parent.L1Origin, n.L1Origin)\r\n            }\r\n            // sanity check that the later sequence number is 0, if it changed between the L2 blocks\r\n            if n.SequenceNumber != 0 {\r\n               return nil, fmt.Errorf(\"l2 block %s has parent %s with different L1 origin %s, but non-zero sequence number %d\", n, parent, parent.L1Origin, n.SequenceNumber)\r\n            }\r\n            // if the L1 origin is known to be canonical, then the parent must be too\r\n            if l1Block.Hash == n.L1Origin.Hash && l1Block.ParentHash != parent.L1Origin.Hash {\r\n               return nil, fmt.Errorf(\"parent L2 block %s has origin %s but expected %s\", parent, parent.L1Origin, l1Block.ParentHash)\r\n            }\r\n         } else {\r\n            if parent.SequenceNumber+1 != n.SequenceNumber {\r\n               return nil, fmt.Errorf(\"sequence number inconsistency %d <> %d between l2 blocks %s and %s\", parent.SequenceNumber, n.SequenceNumber, parent, n)\r\n            }\r\n         }\r\n\r\n         n = parent\r\n\r\n         // once we found the block at seq nr 0 that is more than a full seq window behind the common chain post-reorg, then use the parent block as safe head.\r\n         if ready {\r\n            result.Safe = n\r\n            return result, nil\r\n         }\r\n      }\r\n   }\r\n```\r\n\r\n\r\n### op-node RPC 同步\r\n\r\n这种同步场景处于： 当你有信任的l2 rpc节点的时候，我们可以直接和rpc通信，发送较短范围的同步请求，和2类似。如果设置，在反向链同步中会优先使用RPC而不是P2P同步。\r\n\r\n#### 关键代码\r\n\r\n`op-node/node/node.go`\r\n\r\n初始化rpcSync，如果rpcSyncClient设置，赋值给rpcSync\r\n\r\n```go\r\n   func (n *OpNode) initRPCSync(ctx context.Context, cfg *Config) error {\r\n      rpcSyncClient, rpcCfg, err := cfg.L2Sync.Setup(ctx, n.log, &cfg.Rollup)\r\n      if err != nil {\r\n         return fmt.Errorf(\"failed to setup L2 execution-engine RPC client for backup sync: %w\", err)\r\n      }\r\n      if rpcSyncClient == nil { // if no RPC client is configured to sync from, then don't add the RPC sync client\r\n         return nil\r\n      }\r\n      syncClient, err := sources.NewSyncClient(n.OnUnsafeL2Payload, rpcSyncClient, n.log, n.metrics.L2SourceCache, rpcCfg)\r\n      if err != nil {\r\n         return fmt.Errorf(\"failed to create sync client: %w\", err)\r\n      }\r\n      n.rpcSync = syncClient\r\n      return nil\r\n   }\r\n```\r\n\r\n启动node，如果rpcSync非空，开启`rpcSync eventloop`\r\n\r\n```go\r\n   func (n *OpNode) Start(ctx context.Context) error {\r\n      n.log.Info(\"Starting execution engine driver\")\r\n\r\n      // start driving engine: sync blocks by deriving them from L1 and driving them into the engine\r\n      if err := n.l2Driver.Start(); err != nil {\r\n         n.log.Error(\"Could not start a rollup node\", \"err\", err)\r\n         return err\r\n      }\r\n\r\n      // If the backup unsafe sync client is enabled, start its event loop\r\n      if n.rpcSync != nil {\r\n         if err := n.rpcSync.Start(); err != nil {\r\n            n.log.Error(\"Could not start the backup sync client\", \"err\", err)\r\n            return err\r\n         }\r\n         n.log.Info(\"Started L2-RPC sync service\")\r\n      }\r\n\r\n      return nil\r\n   }\r\n```\r\n\r\n`op-node/sources/sync_client.go`\r\n\r\n一旦接收到`s.requests`通道里的信号后（区块号），调用`fetchUnsafeBlockFromRpc`函数从RPC节点中获取相应的区块信息。\r\n\r\n```go\r\n   // eventLoop is the main event loop for the sync client.\r\n   func (s *SyncClient) eventLoop() {\r\n      defer s.wg.Done()\r\n      s.log.Info(\"Starting sync client event loop\")\r\n\r\n      backoffStrategy := &retry.ExponentialStrategy{\r\n         Min:       1000 * time.Millisecond,\r\n         Max:       20_000 * time.Millisecond,\r\n         MaxJitter: 250 * time.Millisecond,\r\n      }\r\n\r\n      for {\r\n         select {\r\n         case <-s.resCtx.Done():\r\n            s.log.Debug(\"Shutting down RPC sync worker\")\r\n            return\r\n         case reqNum := <-s.requests:\r\n            _, err := retry.Do(s.resCtx, 5, backoffStrategy, func() (interface{}, error) {\r\n               // Limit the maximum time for fetching payloads\r\n               ctx, cancel := context.WithTimeout(s.resCtx, time.Second*10)\r\n               defer cancel()\r\n               // We are only fetching one block at a time here.\r\n               return nil, s.fetchUnsafeBlockFromRpc(ctx, reqNum)\r\n            })\r\n            if err != nil {\r\n               if err == s.resCtx.Err() {\r\n                  return\r\n               }\r\n               s.log.Error(\"failed syncing L2 block via RPC\", \"err\", err, \"num\", reqNum)\r\n               // Reschedule at end of queue\r\n               select {\r\n               case s.requests <- reqNum:\r\n               default:\r\n                  // drop syncing job if we are too busy with sync jobs already.\r\n               }\r\n            }\r\n         }\r\n      }\r\n   }\r\n```\r\n\r\n接下来我们来看看从哪里往`s.requests`通道发送信号的呢？\r\n同文件下的`RequestL2Range`函数，此函数介绍一个需要同步的区块范围，然后将任务通过for循环，分别发送出去。\r\n```go\r\n   func (s *SyncClient) RequestL2Range(ctx context.Context, start, end eth.L2BlockRef) error {\r\n      // Drain previous requests now that we have new information\r\n      for len(s.requests) > 0 {\r\n         select { // in case requests is being read at the same time, don't block on draining it.\r\n         case <-s.requests:\r\n         default:\r\n            break\r\n         }\r\n      }\r\n\r\n      endNum := end.Number\r\n      if end == (eth.L2BlockRef{}) {\r\n         n, err := s.rollupCfg.TargetBlockNumber(uint64(time.Now().Unix()))\r\n         if err != nil {\r\n            return err\r\n         }\r\n         if n <= start.Number {\r\n            return nil\r\n         }\r\n         endNum = n\r\n      }\r\n\r\n      // TODO(CLI-3635): optimize the by-range fetching with the Engine API payloads-by-range method.\r\n\r\n      s.log.Info(\"Scheduling to fetch trailing missing payloads from backup RPC\", \"start\", start, \"end\", endNum, \"size\", endNum-start.Number-1)\r\n\r\n      for i := start.Number + 1; i < endNum; i++ {\r\n         select {\r\n         case s.requests <- i:\r\n         case <-ctx.Done():\r\n            return ctx.Err()\r\n         }\r\n      }\r\n      return nil\r\n   }\r\n```\r\n\r\n在外层的OpNode类型的`RequestL2Range`实现方法里。可以清楚的看到`rpcSync`类型的反向链同步是优先选择的。\r\n\r\n```go\r\n   func (n *OpNode) RequestL2Range(ctx context.Context, start, end eth.L2BlockRef) error {\r\n      if n.rpcSync != nil {\r\n         return n.rpcSync.RequestL2Range(ctx, start, end)\r\n      }\r\n      if n.p2pNode != nil && n.p2pNode.AltSyncEnabled() {\r\n         if unixTimeStale(start.Time, 12*time.Hour) {\r\n            n.log.Debug(\"ignoring request to sync L2 range, timestamp is too old for p2p\", \"start\", start, \"end\", end, \"start_time\", start.Time)\r\n            return nil\r\n         }\r\n         return n.p2pNode.RequestL2Range(ctx, start, end)\r\n      }\r\n      n.log.Debug(\"ignoring request to sync L2 range, no sync method available\", \"start\", start, \"end\", end)\r\n      return nil\r\n   }\r\n```\r\n## 总结\r\n理解了这些同步方式后，我们知道了`unsafe的payload`（区块）究竟是怎么进行传递的。不同的sync模块对应着在不同场景下的区块数据传递。那么整个网络中如何一步步的将`unsafe`的区块变成`safe`区块，然后再进行finalized的呢？这些内容会在其他章节进行讲解。\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":"QmVx3qb6bMYk9LnNkTT18Gk6bRABcg4QHDUpqQGEM5PQLv","timestamp":1698110906,"version":1}