Verification on causuality provided by Lamport timestamp

lamport_timestamp.js

const cluster = require('cluster')
const N = 8

if(cluster.isMaster){
  let workers = {}
  let globals = []
  for(let i=0; i<N; i++) workers[i] = cluster.fork({id: i});
  cluster.on('message', (worker, {message, timestamp, to}) => {
    globals.push({timestamp, message})
    globals.sort(function(x,y){
      let a = x.timestamp
      let b = y.timestamp
      return a[0] - b[0] || a[1] - b[1];
    })
    // console.log(timestamp)
    workers[to].send({timestamp, message, globals})
  })
} else {
  let locals = []
  let clock = 0
  let id = parseInt(process.env.id)

  function verify(globals){
    for(let i=0; i<locals.length; i++){
      // only consider those events made by this process
      if(locals[i].timestamp[1] !== id) continue
      let ixi = globals.findIndex(x => x.message === locals[i].message);
      if(ixi === -1) continue // not yet reach master  

      // global casuality must match local casuality
      for(let j=0; j<i; j++){
        let ixj = globals.findIndex(x => x.message === locals[j].message);
        if(ixi < ixj) return [locals[i], locals[j]] 
      }
    }
    return null 
  }

  process.on('message', ({timestamp, message, globals}) => {
    // lamport timestamp 
    clock = Math.max(clock+1, timestamp[0])
    console.log(`worker ${id} receive ${message} at clock ${clock}`)

    locals.push({timestamp, message})
    if(verify(globals)){
      console.log(id)
      console.log(verify(globals))
      console.log(JSON.stringify(globals))
    }
  })

  setInterval(_ => {
    let timestamp = [++clock, id] 
    let message = `message from ${id} with timestamp ${timestamp}`

    // send to non-self
    let to = id
    while(to === id) to = Math.floor(Math.random() * N);

    locals.push({timestamp, message})
    process.send({timestamp, message, to})
  }, 100 + Math.random() * 200)
}