type MessageWithSender struct {
    Msg    Message // 这是一个接口类型
    Sender commontypes.OracleID
}
type Message interface {
    process(o *oracleState, sender commontypes.OracleID) // 都需要实现 process 这个方法
}

func (o *oracleState) run() {
     
    ......
 
    for {
        select { // 监听和管理P2P消息处理流程
        case msg := <-chNet:
            if 0 <= int(msg.Sender) && int(msg.Sender) < o.Config.N() {
                msg.Msg.process(o, msg.Sender) // 这里本质上是调用 message 自身的 process 方法，将消息转发到相应的信道当中去
            } else {
                o.logger.Critical("msg.Sender out of bounds. This should *never* happen.", commontypes.LogFields{
                    "sender": msg.Sender,
                    "n":      o.Config.N(),
                })
            }
        case epoch := <-chPacemakerToOracle:
            o.epochChanged(epoch)
        case <-chDone:
        }
        ......
    }
}

func (msg MessageNewEpoch) process(o *oracleState, sender commontypes.OracleID) {
    o.chNetToPacemaker <- MessageToPacemakerWithSender{msg, sender}
}

func (msg ...) process(o *oracleState, sender commontypes.OracleID) {
   o.reportGenerationMessage(msg, sender)
}
  
func (o *oracleState) reportGenerationMessage(msg MessageToReportGeneration, sender commontypes.OracleID) {
   msgEpoch := msg.epoch()
   if msgEpoch < o.epoch { // past 消息
      // drop
      o.logger.Debug("oracle: dropping message for past epoch", commontypes.LogFields{
         "epoch":  o.epoch,
         "sender": sender,
         "msg":    msg,
      })
   } else if msgEpoch == o.epoch { // current msg
      o.chNetToReportGeneration <- MessageToReportGenerationWithSender{msg, sender}
   } else { // future msg
      o.bufferedMessages[sender].Push(msg)
      o.logger.Trace("oracle: buffering message for future epoch", commontypes.LogFields{
         "epoch":  o.epoch,
         "sender": sender,
         "msg":    msg,
      })
   }
}

func (repgen *reportGenerationState) messageObserveReq(msg MessageObserveReq, sender commontypes.OracleID) {
     
    // 检查 message validation
    dropPrefix := "messageObserveReq: dropping MessageObserveReq from "
 
    if msg.Epoch != repgen.e {
        repgen.logger.Debug(dropPrefix+"wrong epoch",
            commontypes.LogFields{"round": repgen.followerState.r, "msgEpoch": msg.Epoch},
        )
        return
    }
    if sender != repgen.l {
        repgen.logger.Warn(dropPrefix+"non-leader",
            commontypes.LogFields{"round": repgen.followerState.r, "sender": sender})
        return
    }
    if msg.Round <= repgen.followerState.r {
        repgen.logger.Debug(dropPrefix+"earlier round",
            commontypes.LogFields{"round": repgen.followerState.r, "msgRound": msg.Round})
        return
    }
    if int64(repgen.config.RMax)+1 < int64(msg.Round) {
        repgen.logger.Warn(dropPrefix+"out of bounds round",
            commontypes.LogFields{"round": repgen.followerState.r, "rMax": repgen.config.RMax, "msgRound": msg.Round})
        return
    }
 
    // 更新 followerState 中记录的当前 round
    repgen.followerState.r = msg.Round
 
    // 判断时候发起 epoch change
    if repgen.followerState.r > repgen.config.RMax {
        repgen.logger.Debug(
            "messageReportReq: leader sent MessageObserveReq past its expiration "+
                "round. Time to change leader",
            commontypes.LogFields{
                "round":        repgen.followerState.r,
                "messageRound": msg.Round,
                "roundMax":     repgen.config.RMax,
            })
        select {
        case repgen.chReportGenerationToPacemaker <- EventChangeLeader{}:
        case <-repgen.ctx.Done():
        }
 
        return
    }
     
    // 初始化 followerState
    repgen.followerState.sentEcho = nil
    repgen.followerState.sentReport = false
    repgen.followerState.completedRound = false
    repgen.followerState.receivedEcho = make([]bool, repgen.config.N())
 
    repgen.telemetrySender.RoundStarted(
        repgen.config.ConfigDigest,
        repgen.e,
        repgen.followerState.r,
        repgen.l,
    )
 
    // 获取 observation
    value := repgen.observeValue()
 
    ......
 
    // 签名
    so, err := MakeSignedObservation(value, repgen.followerReportContext(), repgen.privateKeys.SignOffChain)
 
    ......
 
    // 发送 MessageObservation 给 leader
    repgen.netSender.SendTo(MessageObserve{
        repgen.e,
        repgen.followerState.r,
        so,
    }, repgen.l)
}

func (repgen *reportGenerationState) run() {
    for {
        select {
        ......
        case <-repgen.leaderState.tGrace: // tGrace 计时结束，要去发 MessageReport 了
            repgen.eventTGraceTimeout()
        }
    }
}

func (repgen *reportGenerationState) eventTGraceTimeout() {
    // 验证当前状态
    if repgen.leaderState.phase != phaseGrace {
        repgen.logger.Error("leader's phase conflicts tGrace timeout", commontypes.LogFields{
            "round": repgen.leaderState.r,
            "phase": englishPhase[repgen.leaderState.phase],
        })
        return
    }
    // 构建 signed observation 集
    asos := []AttributedSignedObservation{}
    for oid, so := range repgen.leaderState.observe {
        if so != nil {
            asos = append(asos, AttributedSignedObservation{
                *so,
                commontypes.OracleID(oid),
            })
        }
    }
    // observation从小到大排序
    sort.Slice(asos, func(i, j int) bool {
        return asos[i].SignedObservation.Observation.Less(asos[j].SignedObservation.Observation)
    })
    // 广播 MessageReportReq
    repgen.netSender.Broadcast(MessageReportReq{
        repgen.e,
        repgen.leaderState.r,
        asos,
    })
    // 当前 round 的 state 移动到 pharseReport
    repgen.leaderState.phase = phaseReport
}

func (pace *pacemakerState) messageNewepoch(msg MessageNewEpoch, sender commontypes.OracleID) {
    // 检查sender的id范围
    if int(sender) < 0 || int(sender) >= len(pace.newepoch) {
        pace.logger.Error("Pacemaker: dropping NewEpoch message from invalid sender", commontypes.LogFields{
            "sender": sender,
            "N":      len(pace.newepoch),
        })
        return
    }
    // 该 sender 发过来的最高 epoch < 这个消息的 epoch
    if pace.newepoch[sender] < msg.Epoch {
        pace.newepoch[sender] = msg.Epoch // 更新
        pace.persist() // 持久化
    } else {
        // 说明是 past epoch 不用处理了
        return
    }
 
    // upon |{p_j ∈ P | newepoch[j] > ne}| > f do
    {   // 这里的功能是，计算我这个节点是否需要向p2p网络中广播newepoch消息
        // pace.ne记录的是我发出的newepoch消息中最高的epoch
        // 如果我收到的周围节点发给我的最高epoch比我高，且人数达到了 F 。
        // 我就需要发出newepoch消息，并更新我的pace.ne
        // 至于我发出的newepoch的epoch number是多少，则选取我收到的满足 F 人的前提下最高的epoch number
        candidateEpochs := sortedGreaterThan(pace.newepoch, pace.ne)
        if len(candidateEpochs) > pace.config.F {
            // ē ← max {e' | {p_j ∈ P | newepoch[j] ≥ e' } > f}
            newEpoch := candidateEpochs[len(candidateEpochs)-(pace.config.F+1)]
            pace.sendNewepoch(newEpoch) // 更新并持久化pace.ne，然后广播newepoch消息
        }
    }
 
    // upon |{p_j ∈ P | newepoch[j] > e}| > 2f do
    {
        // pace.e是当前所处的epoch(即最高的完成聚合共识的epoch)
        candidateEpochs := sortedGreaterThan(pace.newepoch, pace.e)
        if len(candidateEpochs) > 2*pace.config.F {
            // ē ← max {e' | {p_j ∈ P | newepoch[j] ≥ e' } > 2f}
            //
            // since candidateEpochs contains, in increasing order, the epochs from
            // the received newepoch messages, this value of newEpoch was sent by at
            // least 2F+1 processes
            // 2f+1的人都同意的epoch，这个epoch算达成共识了
            newEpoch := candidateEpochs[len(candidateEpochs)-(2*pace.config.F+1)]
            pace.logger.Debug("Moving to epoch, based on candidateEpochs", commontypes.LogFields{
                "newEpoch":        newEpoch,
                "candidateEpochs": candidateEpochs,
            })
            // 计算 leader 的 id 。
            // 将 (epoch number, 共享密钥) 组合后进行hash后求和再根据总人数N取余
            l := Leader(newEpoch, pace.config.N(), pace.config.LeaderSelectionKey())
            // 更新pace.e(达成共识的最高epoch)和pace.l(最高共识epoch的leader)
            pace.e, pace.l = newEpoch, l // (e, l) ← (ē, leader(ē))
            if pace.ne < pace.e {        // ne ← max{ne, e}
                pace.ne = pace.e
            }
            pace.persist() // 持久化
 
            // abort instance [...], initialize instance (e,l) of report generation
            pace.spawnReportGeneration() // 生成 reportGeneration
 
            // 通知 NewEpoch 产生了
            pace.notifyOracleOfNewEpoch = true
 
            // 重置超时计时
            pace.tProgress = time.After(pace.config.DeltaProgress) // restart timer T_{progress}
        }
    }
}

func (o *oracleState) run() {
    ......
    for {
        select {
        ......
        case epoch := <-chPacemakerToOracle: // epoch更新后，会进入到这下面
            o.epochChanged(epoch)
        ......
        }
    }
    ......
}
  
func (o *oracleState) epochChanged(e uint32) {
   // 清理缓存中的消息
   o.epoch = e
   o.logger.Trace("epochChanged: getting messages for new epoch", commontypes.LogFields{
      "epoch": e,
   })
   // id其实就是sender的id，buffer就是这个sender发过来的所有MessageToReportGeneration
   for id, buffer := range o.bufferedMessages {
      for {
         msg := buffer.Peek() // 取出 MessageToReportGeneration
         if msg == nil {
            // no messages left in buffer
            break
         }
         msgEpoch := (*msg).epoch() // 获取这个 msg 的 epoch
         if msgEpoch < e {
            // 我们当前变更到的epoch要低，就是past msg
            // 从 buffer 移出就行了，不用管
            buffer.Pop()
            o.logger.Debug("epochChanged: unbuffered and dropped message", commontypes.LogFields{
               "remoteOracleID": id,
               "epoch":          e,
               "message":        *msg,
            })
         } else if msgEpoch == e {
            // 如果这是我这个epoch的msg，说明是current msg需要处理
            // remove from buffer
            buffer.Pop()
 
            o.logger.Trace("epochChanged: unbuffered messages for new epoch", commontypes.LogFields{
               "remoteOracleID": id,
               "epoch":          e,
               "message":        *msg,
            })
            // 将消息转发到 chNetToReportGeneration channal 中
            o.chNetToReportGeneration <- MessageToReportGenerationWithSender{
               *msg,
               commontypes.OracleID(id),
            }
         } else {
            // 说明依旧还是future msg，保持原样
            // this and all subsequent messages are for future epochs
            // leave them in the buffer
            break
         }
      }
   }
   o.logger.Trace("epochChanged: done getting messages for new epoch", commontypes.LogFields{
      "epoch": e,
   })
}

// 收到的 MessageNewEpoch 满足了 2f+1 要求，视图切换确认
if len(candidateEpochs) > 2*pace.config.F {
......
    pace.spawnReportGeneration() // 重新初始化一个新的 reportGenerationState 模块，用于该新视图中的聚合轮次处理
    ......
}

func Leader(epoch uint32, n int, key [16]byte) (leader commontypes.OracleID) {
   // No need for HMAC. Since we use Keccak256, prepending
   // with key gives us a PRF already.
   h := sha3.NewLegacyKeccak256()
   h.Write(key[:])
   b := make([]byte, 8)
   binary.LittleEndian.PutUint64(b, uint64(epoch))
   h.Write(b)
 
   result := big.NewInt(0)
   r := big.NewInt(0).SetBytes(h.Sum(nil))
   // This is biased, but we don't care because the prob of us hitting the bias are
   // less than 2**5/2**256 = 2**-251.
   result.Mod(r, big.NewInt(int64(n)))
   return commontypes.OracleID(result.Int64())
}