epoch-voters.js

/* eslint-env mocha */
'use strict'

const chai = require('chai')
const dirtyChai = require('dirty-chai')
const expect = chai.expect
chai.use(dirtyChai)

const CRDT = require('../')
const transmit = require('./transmit')

const EpochVotersCRDT = {
  initial: () => [1, new Map()],
  join: (s1, s2) => {
    const epoch = Math.max(s1[0], s2[0])
    const res = [epoch]
    if (s1[0] > s2[0]) {
      res[1] = new Map(s1[1])
    } else if (s2[0] > s1[0]) {
      res[1] = new Map(s2[1])
    } else {
      res[1] = new Map([...s1[1], ...s2[1]])
    }
    return res
  },
  value: (state) => {
    let leader = null
    const countByChoice = [...state[1]].reduce((acc, [voter, choice]) => {
      acc.set(choice, (acc.get(choice) || 0) + 1)
      return acc
    }, new Map())
    const sorted = [...countByChoice].sort((a, b) => b[1] - a[1])
    if (sorted.length && sorted[0][1] !== (sorted[1] || [])[1]) {
      leader = sorted[0][0]
    }
    return [leader, state[1], state[0]]
  },
  mutators: {
    vote (id, state, choice) {
      return [state[0], new Map([[id, choice]])]
    },
    voteNewEpoch (id, state, choice) {
      return [state[0] + 1, new Map([[id, choice]])]
    }
  }
}

CRDT.define('epochvoters', EpochVotersCRDT)

describe('epoch voters', () => {
  describe('local', () => {
    let EpochVoters
    let epochVoters
    it('type can be created', () => {
      EpochVoters = CRDT('epochvoters')
    })

    it('can be instantiated', () => {
      epochVoters = EpochVoters('id1')
    })

    it('starts empty at epoch 1', () => {
      expect(epochVoters.value()).to.deep.equal([null, new Map(), 1])
    })

    it('can vote', () => {
      epochVoters.vote('someid')
    })

    it('and the vote is recorded', () => {
      expect(epochVoters.value()).to.deep.equal(['someid', obj2Map({'id1': 'someid'}), 1])
    })

    it('can advance the epoch and vote', () => {
      epochVoters.voteNewEpoch('someotherid')
    })

    it('and the vote is recorded and epoch incremented', () => {
      expect(epochVoters.value()).to.deep.equal(['someotherid', obj2Map({'id1': 'someotherid'}), 2])
    })
  })

  describe('convergence', () => {
    let EpochVoters = CRDT('epochvoters')

    let replica1, replica2, replica3
    let deltas = [[], [], []]
    before(() => {
      replica1 = EpochVoters('id1')
      replica2 = EpochVoters('id2')
      replica3 = EpochVoters('id3')
    })

    it('can vote', () => {
      deltas[0].push(replica1.vote('apples'))
      deltas[1].push(replica2.vote('oranges'))
      deltas[2].push(replica3.vote('apples'))
    })

    it('changes from others can be joined to the first', () => {
      deltas[1].forEach((delta) => replica1.apply(transmit(delta)))
      deltas[2].forEach((delta) => replica1.apply(transmit(delta)))
    })

    it('changes from others can be joined to the second', () => {
      deltas[0].forEach((delta) => replica2.apply(transmit(delta)))
      deltas[2].forEach((delta) => replica2.apply(transmit(delta)))
    })

    it('changes from others can be joined to the third', () => {
      deltas[0].forEach((delta) => replica3.apply(transmit(delta)))
      deltas[1].forEach((delta) => replica3.apply(transmit(delta)))
    })

    const exp = {'id1': 'apples', 'id2': 'oranges', 'id3': 'apples'}
    it('the first converges', () => {
      expect(replica1.value()).to.deep.equal(['apples', obj2Map(exp), 1])
    })

    it('and the second also converges', () => {
      expect(replica2.value()).to.deep.equal(['apples', obj2Map(exp), 1])
    })

    it('and the third also converges', () => {
      expect(replica3.value()).to.deep.equal(['apples', obj2Map(exp), 1])
    })
  })

  describe('no leader', () => {
    let EpochVoters = CRDT('epochvoters')

    let replica1, replica2, replica3
    let deltas = [[], [], []]
    before(() => {
      replica1 = EpochVoters('id1')
      replica2 = EpochVoters('id2')
      replica3 = EpochVoters('id3')
    })

    it('first two can vote', () => {
      deltas[0].push(replica1.vote('apples'))
      deltas[1].push(replica2.vote('oranges'))
    })

    it('changes from second can be joined to the first', () => {
      deltas[1].forEach((delta) => replica1.apply(transmit(delta)))
    })

    it('changes from first can be joined to the second', () => {
      deltas[0].forEach((delta) => replica2.apply(transmit(delta)))
    })

    it('there is no leader in the first', () => {
      expect(replica1.value()[0]).to.be.null()
    })

    it('there is no leader in the second', () => {
      expect(replica1.value()[0]).to.be.null()
    })

    it('third vote is tie-breaker', () => {
      deltas[2].push(replica3.vote('apples'))
      deltas[2].forEach((delta) => replica1.apply(transmit(delta)))
      deltas[2].forEach((delta) => replica2.apply(transmit(delta)))
      expect(replica1.value()[0]).to.equal('apples')
      expect(replica2.value()[0]).to.equal('apples')
    })
  })

  describe('epoch change', () => {
    let EpochVoters = CRDT('epochvoters')

    let replica1, replica2, replica3
    before(() => {
      replica1 = EpochVoters('id1')
      replica2 = EpochVoters('id2')
      replica3 = EpochVoters('id3')
    })

    it('after join replicas converge', () => {
      const deltas = [[], [], []]
      deltas[0].push(replica1.vote('apples'))
      deltas[1].push(replica2.vote('oranges'))
      deltas[2].push(replica3.vote('oranges'))

      deltas[1].forEach((delta) => replica1.apply(transmit(delta)))
      deltas[2].forEach((delta) => replica1.apply(transmit(delta)))
      deltas[0].forEach((delta) => replica2.apply(transmit(delta)))
      deltas[2].forEach((delta) => replica2.apply(transmit(delta)))
      deltas[0].forEach((delta) => replica3.apply(transmit(delta)))
      deltas[1].forEach((delta) => replica3.apply(transmit(delta)))

      const exp = {'id1': 'apples', 'id2': 'oranges', 'id3': 'oranges'}
      expect(replica1.value()).to.deep.equal(['oranges', obj2Map(exp), 1])
      expect(replica2.value()).to.deep.equal(['oranges', obj2Map(exp), 1])
      expect(replica3.value()).to.deep.equal(['oranges', obj2Map(exp), 1])
    })

    it('when one replica moves to next epoch, after join replicas ignore previous epoch', () => {
      const deltas = [[], [], []]
      deltas[0].push(replica1.voteNewEpoch('oranges'))

      deltas[0].forEach((delta) => replica2.apply(transmit(delta)))
      deltas[0].forEach((delta) => replica3.apply(transmit(delta)))

      const exp = {'id1': 'oranges'}
      expect(replica1.value()).to.deep.equal(['oranges', obj2Map(exp), 2])
      expect(replica2.value()).to.deep.equal(['oranges', obj2Map(exp), 2])
      expect(replica3.value()).to.deep.equal(['oranges', obj2Map(exp), 2])
    })

    it('subsequent votes occur in the new epoch', () => {
      const deltas = [[], [], []]
      deltas[1].push(replica2.vote('apples'))
      deltas[2].push(replica3.vote('apples'))

      deltas[1].forEach((delta) => replica1.apply(transmit(delta)))
      deltas[2].forEach((delta) => replica1.apply(transmit(delta)))
      deltas[1].forEach((delta) => replica3.apply(transmit(delta)))
      deltas[2].forEach((delta) => replica2.apply(transmit(delta)))

      const exp = {'id1': 'oranges', 'id2': 'apples', 'id3': 'apples'}
      expect(replica1.value()).to.deep.equal(['apples', obj2Map(exp), 2])
      expect(replica2.value()).to.deep.equal(['apples', obj2Map(exp), 2])
      expect(replica3.value()).to.deep.equal(['apples', obj2Map(exp), 2])
    })
  })
})

function obj2Map(obj) {
  return new Map(Object.entries(obj))
}