Membership and failure

// mergeTable keeps the higher heartbeat counter for each node. A counter that grew proves the
// node was alive recently, so a node marked dead comes back.
func (s *sim) mergeTable(i int, table []int64) {
  nd := s.nodes[i]
  for j, b := range table {
    if j != i && b > nd.view[j].beat {
      nd.view[j].beat, nd.view[j].fresh = b, s.now
      s.setState(i, j, alive)
    }
  }
}

// check marks dead every node whose counter has not grown for Timeout.
func (s *sim) check(i int) {
  nd := s.nodes[i]
  if !nd.up {
    return
  }
  s.at(s.now+50, func() { s.check(i) })
  for j, m := range nd.view {
    if j != i && m.st == alive && s.now-m.fresh > s.cfg.Timeout {
      s.setState(i, j, dead)
    }
  }
}

// Phi is the phi accrual detector: it keeps the recent gaps between heartbeats and turns the
// silence since the last one into a suspicion level. phi = 8 means the chance that a heartbeat
// is still on its way is 10^-8.
type Phi struct {
  gaps   []float64
  max    int
  minStd float64
  last   float64
}

// NewPhi keeps up to max gaps, and never lets the spread fall below minStd ms.
func NewPhi(max int, minStd float64) *Phi { return &Phi{max: max, minStd: minStd, last: -1} }

// Heartbeat records an arrival at time t.
func (p *Phi) Heartbeat(t float64) {
  if p.last >= 0 {
    p.gaps = append(p.gaps, t-p.last)
    if len(p.gaps) > p.max {
      p.gaps = p.gaps[1:]
    }
  }
  p.last = t
}

func (p *Phi) stats() (mean, std float64) {
  for _, g := range p.gaps {
    mean += g
  }
  mean /= float64(len(p.gaps))
  for _, g := range p.gaps {
    std += (g - mean) * (g - mean)
  }
  return mean, max(math.Sqrt(std/float64(len(p.gaps))), p.minStd)
}

// Level is phi at time now: -log10 of the chance that the gap grows this long, for a normal
// distribution fitted to the recent gaps.
func (p *Phi) Level(now float64) float64 {
  mean, std := p.stats()
  later := 0.5 * math.Erfc((now-p.last-mean)/(std*math.Sqrt2))
  return -math.Log10(later)
}

// After is how long after the last heartbeat phi reaches threshold.
func (p *Phi) After(threshold float64) float64 {
  mean, std := p.stats()
  return mean + std*math.Sqrt2*math.Erfcinv(2*math.Pow(10, -threshold))
}

// probe runs once per Period on node i: ping one member, then ask K others, then suspect.
func (s *sim) probe(i int) {
  nd := s.nodes[i]
  if !nd.up {
    return
  }
  s.at(s.now+s.cfg.Period, func() { s.probe(i) })
  t := s.nextTarget(i)
  if t < 0 {
    return
  }
  acked := false
  ack := func() { acked = true }
  s.send(i, t, 0, func() { s.send(t, i, 0, ack) }) // ping, ack
  s.at(s.now+s.cfg.ProbeTimeout, func() {
    if acked || !nd.up {
      return
    }
    for _, h := range s.peers(i, s.cfg.K, t, false) { // indirect: h pings t for i
      s.send(i, h, 0, func() {
        s.send(h, t, 0, func() {
          s.send(t, h, 0, func() { s.send(h, i, 0, ack) })
        })
      })
    }
  })
  s.at(s.now+s.cfg.Period-1, func() {
    if !acked && nd.up {
      s.apply(i, update{st: suspect, node: t, inc: nd.view[t].inc})
    }
  })
}

// apply handles one membership update at node i: a local finding or a gossiped one.
func (s *sim) apply(i int, u update) {
  nd := s.nodes[i]
  if u.node == i {
    if u.st != alive && u.inc >= nd.inc {
      nd.inc = u.inc + 1 // refute: I am alive, with a higher incarnation
      s.enqueue(i, update{st: alive, node: i, inc: nd.inc})
    }
    return
  }
  m := &nd.view[u.node]
  switch u.st {
  case alive:
    if u.inc <= m.inc {
      return
    }
    m.inc = u.inc
    m.token++
    s.setState(i, u.node, alive)
  case suspect:
    if m.st == dead || u.inc < m.inc || (m.st == suspect && u.inc == m.inc) {
      return
    }
    m.inc = u.inc
    m.token++
    tok := m.token
    s.setState(i, u.node, suspect)
    s.at(s.now+s.cfg.Timeout, func() {
      if nd.up && m.st == suspect && m.token == tok {
        s.apply(i, update{st: dead, node: u.node, inc: m.inc})
      }
    })
  case dead:
    if m.st == dead || u.inc < m.inc {
      return
    }
    m.inc = u.inc
    s.setState(i, u.node, dead)
  }
  s.enqueue(i, u)
}