<commit_before>var assert = require('assert')
var Monotonic = require('monotonic').asString
var Scheduler = require('happenstance')
var push = [].push
var slice = [].slice
var RBTree = require('bintrees').RBTree
var logger = require('prolific.logger').createLogger('bigeasy.paxos')

function Legislator (islandId, id, cookie, options) {
    options || (options = {})

    this.islandId = islandId
    this.id = id
    this.cookie = cookie
    this.naturalized = !! options.naturalized

    this.parliamentSize = options.parliamentSize || 5

    this.log = new RBTree(function (a, b) { return Monotonic.compare(a.promise, b.promise) })
    this.scheduler = new Scheduler(options.scheduler || {})
    this.synchronizing = {}

    this.proposals = []
    this.citizens = {}
    this.keepAlive = false
    this.immigrating = []
    this.collapsed = false

    this.government = { promise: '0/0', minority: [], majority: [] }
    this.promise = '0/0'

    this._peers = {}
    this.getPeer(this.id).timeout = 0

    this.length = options.length || 1024

// Randomly adjust election retry by a percentage.

    this.ping = options.ping || 1
    this.timeout = options.timeout || 3

    this.log.insert({
        promise: '0/0',
        value: { government: this.government },
        quorum: [ this.id ],
        decisions: [ this.id ],
        decided: true
    })

    this.constituency = []
}

function trace (method, vargs) {
    logger.trace(method, { vargs: vargs })
}

Legislator.prototype.getPeer = function (id) {
    trace('getPeer', [ id ])
    var peer = this._peers[id]
    if (peer == null) {
        return peer = this._peers[id] = {
            timeout: 0,
            when: null,
            pinged: false,
            decided: '0/0'
        }
    }
    return peer
}

Legislator.prototype.newGovernment = function (now, quorum, government, promise) {
    trace('newGovernment', [ now, quorum, government, promise ])
    assert(!government.constituents)
    government.constituents = Object.keys(this.citizens).filter(function (citizen) {
        return !~government.majority.indexOf(citizen)
            && !~government.minority.indexOf(citizen)
    })
    var remapped = government.promise = promise
    this.proposals = this.proposals.splice(0, this.proposals.length).map(function (proposal) {
        proposal.was = proposal.promise
        proposal.route = government.majority
        proposal.promise = remapped = Monotonic.increment(remapped, 1)
        return proposal
    }.bind(this))
    var citizens = JSON.parse(JSON.stringify(this.citizens))
// TODO I'd rather have a more intelligent structure.
    if (government.immigrate) {
        citizens[government.immigrate.id] = JSON.parse(JSON.stringify(government.immigrate.properties))
        citizens[government.immigrate.id].immigrated = promise
        government.constituents.push(government.immigrate.id)
    }
    this.proposals.unshift({
        promise: promise,
        route: quorum,
        value: {
            type: 'government',
            islandId: this.islandId,
// TODO Choke up on this structure, move majority and minority up one.
            government: government,
            citizens: citizens,
// TODO Null map to indicate collapse or change in leadership. Wait, change in
// leaership is only ever collapse? Ergo...
            collapsed: this.collapsed,
// There was a time when I wanted to allow the user to choose leaders.
            map: this.proposals.map(function (proposal) {
                return { was: proposal.was, is: proposal.promise }
            })
        }
    })
}

Legislator.prototype._consensus = function (now) {
    trace('consensus', [ now ])
    // Shift any immigrating citizens that have already immigrated.
    while (
        this.immigrating.length != 0 &&
        this.citizens[this.immigrating[0].id]
    ) {
        this.immigrating.shift()
    }
    if (this.collapsed) {
        if (this.election) {
// TODO Currently, your tests are running all synchronizations to completion
// before running a consensus pulse, so we're not seeing the results of decided
// upon a consensus action before all of the synchronizations have been
// returned.
            if (this.election.status == 'accepted') {
                return null
            } else if (this.election.status == 'proposed') {
                if (this.election.accepts.length == this.election.promises.length) {
                    this.election.status = 'accepted'
                    return {
                        type: 'consensus',
                        islandId: this.islandId,
                        governments: [ this.government.promise, this.accepted.promise ],
                        route: this.accepted.route,
                        messages: [this._ping(now), {
                            type: 'commit',
                            promise: this.accepted.promise
                        }]
                    }
                }
                return null
            } else if (this.election.promises.length < this.election.majority.length) {
                return null
            } else {
                this.election.status = 'proposed'
                this.newGovernment(now, this.election.majority, {
                    majority: this.election.majority,
                    minority: this.election.minority
                }, this.promise)
            }
        } else {
            var parliament = this.government.majority.concat(this.government.minority)
// TODO The constituent must be both connected and synchronized, not just
// connected.
            var present = this.parliament.filter(function (id) {
                var peer = this._peers[id] || {}
                return id != this.id && peer.naturalized && peer.timeout == 0
            }.bind(this))
            var majoritySize = Math.ceil(parliament.length / 2)
            if (present.length < majoritySize) {
                return null
            }
            var majority = [ this.id ].concat(present).slice(0, majoritySize)
            var minority = this.parliament.filter(function (id) { return ! ~majority.indexOf(id) })
            this.election = {
                status: 'proposing',
                majority: majority,
                minority: minority,
                promises: [],
                accepts: []
            }
            return {
                type: 'consensus',
                islandId: this.islandId,
                governments: [ this.government.promise ],
                route: majority,
                messages: [{
                    type: 'propose',
                    // Do not increment here, it will be set by `_receivePromise`.
                    promise: Monotonic.increment(this.promise, 0)
                }]
            }
        }
    } else if (this.government.majority[0] == this.id && this.accepted && Monotonic.isBoundary(this.accepted.promise, 0)) {
        return {
            type: 'consensus',
            islandId: this.islandId,
            governments: [ this.government.promise, this.accepted.promise ],
            route: this.accepted.route,
            messages: [this._ping(now), {
                type: 'commit',
                promise: this.accepted.promise
            }]
        }
    } else if (this.immigrating.length && this.government.majority[0] == this.id) {
        var immigration = this.immigrating.shift()
        this.newGovernment(now, this.government.majority, {
            majority: this.government.majority,
            minority: this.government.minority,
            immigrate: {
                id: immigration.id,
                properties: immigration.properties,
                cookie: immigration.cookie
            }
        }, Monotonic.increment(this.promise, 0))
    } else if (this.ponged && this.id == this.government.majority[0]) {
        var reshape = this._impeach() || this._exile() || this._expand()
        if (reshape) {
            this.ponged = false
            this.newGovernment(now, reshape.quorum, reshape.government, Monotonic.increment(this.promise, 0))
        }
    }
    var proposal = this.proposals[0]
    var messages = [ this._ping(now) ]
    if (this.accepted != null) {
        messages.push({
            type: 'commit',
            promise: this.accepted.promise
        })
        if (proposal == null || !this._routeEqual(proposal.route, this.accepted.route)) {
            return {
                type: 'consensus',
                islandId: this.islandId,
                governments: [ this.government.promise ],
                route: this.accepted.route,
                messages: messages
            }
        }
    }
    if (proposal) {
        messages.push({
            type: 'accept',
            promise: proposal.promise,
            value: proposal.value
        })
        this.proposals.shift()
        return {
            type: 'consensus',
            islandId: this.islandId,
            governments: [ this.government.promise ],
            route: proposal.route.slice(),
            messages: messages
        }
    }
    if (this.keepAlive) {
        this.keepAlive = false
        return {
            type: 'consensus',
            islandId: this.islandId,
            governments: [ this.government.promise ],
            route: this.government.majority,
            messages: [ this._ping(now) ]
        }
    }
    return null
}

Legislator.prototype.consensus = function (now) {
    var pulse = null
    if (!this._pulse) {
        pulse = this._consensus(now)
        this._pulse = !! pulse
    }
    return pulse
}


Legislator.prototype.synchronize = function (now) {
    trace('synchronize', [ now ])
    var outbox = []
    for (var i = 0, I = this.constituency.length; i < I; i++) {
        var id = this.constituency[i]
        var peer = this.getPeer(id)
        var compare = Monotonic.compare(this.getPeer(id).decided, this.getPeer(this.id).decided)
        if ((peer.timeout != 0 || compare < 0) && !peer.skip && !this.synchronizing[id]) {
            var count = 20
            this.synchronizing[id] = true
            var pulse = {
                type: 'synchronize',
                islandId: this.islandId,
                governments: [ this.government.promise ],
                route: [ id ],
                messages: []
            }

            var maximum = peer.decided
            if (peer.pinged) {
                var round
                if (peer.decided == '0/0') {
                    var iterator = this.log.iterator()
                    for (;;) {
                        round = iterator.prev()
// TODO This will abend if the naturalization falls off the end end of the log.
// You need to check for gaps and missing naturalizations and then timeout the
// constituents that will never be connected.
                        assert(round, 'cannot find immigration')
                        if (Monotonic.isBoundary(round.promise, 0)) {
                            var immigrate = round.value.government.immigrate
                            if (immigrate && immigrate.id == id) {
                                maximum = round.promise
                                break
                            }
                        }
                    }
                } else {
// TODO Got a read property of null here.
                    round = this.log.find({ promise: maximum }).next
                }

                while (--count && round) {
                    pulse.messages.push({
                        type: 'enact',
                        promise: round.promise,
                        value: round.value
                    })
                    round = round.next
                }
            }

            pulse.messages.push(this._ping(now))

            outbox.push(pulse)
        }
    }
    return outbox
}

Legislator.prototype.receive = function (now, pulse, messages) {
    trace('_receive', [ now, pulse, messages ])
    assert(arguments.length == 3 && now != null)
    var responses = []
    for (var i = 0, I = messages.length; i < I; i++) {
        var message = messages[i]
        var type = message.type
        var method = '_receive' + type[0].toUpperCase() + type.substring(1)
        this[method].call(this, now, pulse, message, responses)
    }
    return responses
}

Legislator.prototype.collapse = function () {
    trace('collapse', [])
    this.collapsed = true
    this.election = null
    this.proposals.length = 0
    this.immigrating.length = 0
    for (var id in this._peers) {
        if (id != this.id) {
            delete this._peers[id]
        }
    }
    var parliament = this.government.majority.concat(this.government.minority)
    this.constituency = parliament.filter(function (id) {
        return this.id != id
    }.bind(this))
}

Legislator.prototype.sent = function (now, pulse, responses) {
    trace('sent', [ now, pulse, responses ])
    if (pulse.type == 'consensus') {
        this._pulse = false
    }
    if (!~pulse.governments.indexOf(this.government.promise)) {
        return
    }
    var success = true
    pulse.route.forEach(function (id) {
        if (responses[id] == null) {
            success = false
        } else {
            this.receive(now, pulse, responses[id])
        }
    }, this)
    success = success && !pulse.failed
    if (success) {
        switch (pulse.type) {
        case 'synchronize':
            delete this.synchronizing[pulse.route[0]]
            this.scheduler.schedule(now + this.ping, pulse.route[0], { object: this, method: '_whenPing' }, pulse.route[0])
            break
        case 'consensus':
            this.scheduler.schedule(now + this.ping, this.id, { object: this, method: '_whenKeepAlive' })<commit_msg>Rename `_pulse` to `pulsing`.<commit_after>var assert = require('assert')
var Monotonic = require('monotonic').asString
var Scheduler = require('happenstance')
var push = [].push
var slice = [].slice
var RBTree = require('bintrees').RBTree
var logger = require('prolific.logger').createLogger('bigeasy.paxos')

function Legislator (islandId, id, cookie, options) {
    options || (options = {})

    this.islandId = islandId
    this.id = id
    this.cookie = cookie
    this.naturalized = !! options.naturalized

    this.parliamentSize = options.parliamentSize || 5

    this.log = new RBTree(function (a, b) { return Monotonic.compare(a.promise, b.promise) })
    this.scheduler = new Scheduler(options.scheduler || {})
    this.synchronizing = {}

    this.proposals = []
    this.citizens = {}
    this.immigrating = []
    this.keepAlive = false
    this.pulsing = false
    this.collapsed = false

    this.government = { promise: '0/0', minority: [], majority: [] }
    this.promise = '0/0'

    this._peers = {}
    this.getPeer(this.id).timeout = 0

    this.length = options.length || 1024

// Randomly adjust election retry by a percentage.

    this.ping = options.ping || 1
    this.timeout = options.timeout || 3

    this.log.insert({
        promise: '0/0',
        value: { government: this.government },
        quorum: [ this.id ],
        decisions: [ this.id ],
        decided: true
    })

    this.constituency = []
}

function trace (method, vargs) {
    logger.trace(method, { vargs: vargs })
}

Legislator.prototype.getPeer = function (id) {
    trace('getPeer', [ id ])
    var peer = this._peers[id]
    if (peer == null) {
        return peer = this._peers[id] = {
            timeout: 0,
            when: null,
            pinged: false,
            decided: '0/0'
        }
    }
    return peer
}

Legislator.prototype.newGovernment = function (now, quorum, government, promise) {
    trace('newGovernment', [ now, quorum, government, promise ])
    assert(!government.constituents)
    government.constituents = Object.keys(this.citizens).filter(function (citizen) {
        return !~government.majority.indexOf(citizen)
            && !~government.minority.indexOf(citizen)
    })
    var remapped = government.promise = promise
    this.proposals = this.proposals.splice(0, this.proposals.length).map(function (proposal) {
        proposal.was = proposal.promise
        proposal.route = government.majority
        proposal.promise = remapped = Monotonic.increment(remapped, 1)
        return proposal
    }.bind(this))
    var citizens = JSON.parse(JSON.stringify(this.citizens))
// TODO I'd rather have a more intelligent structure.
    if (government.immigrate) {
        citizens[government.immigrate.id] = JSON.parse(JSON.stringify(government.immigrate.properties))
        citizens[government.immigrate.id].immigrated = promise
        government.constituents.push(government.immigrate.id)
    }
    this.proposals.unshift({
        promise: promise,
        route: quorum,
        value: {
            type: 'government',
            islandId: this.islandId,
// TODO Choke up on this structure, move majority and minority up one.
            government: government,
            citizens: citizens,
// TODO Null map to indicate collapse or change in leadership. Wait, change in
// leaership is only ever collapse? Ergo...
            collapsed: this.collapsed,
// There was a time when I wanted to allow the user to choose leaders.
            map: this.proposals.map(function (proposal) {
                return { was: proposal.was, is: proposal.promise }
            })
        }
    })
}

Legislator.prototype._consensus = function (now) {
    trace('consensus', [ now ])
    // Shift any immigrating citizens that have already immigrated.
    while (
        this.immigrating.length != 0 &&
        this.citizens[this.immigrating[0].id]
    ) {
        this.immigrating.shift()
    }
    if (this.collapsed) {
        if (this.election) {
// TODO Currently, your tests are running all synchronizations to completion
// before running a consensus pulse, so we're not seeing the results of decided
// upon a consensus action before all of the synchronizations have been
// returned.
            if (this.election.status == 'accepted') {
                return null
            } else if (this.election.status == 'proposed') {
                if (this.election.accepts.length == this.election.promises.length) {
                    this.election.status = 'accepted'
                    return {
                        type: 'consensus',
                        islandId: this.islandId,
                        governments: [ this.government.promise, this.accepted.promise ],
                        route: this.accepted.route,
                        messages: [this._ping(now), {
                            type: 'commit',
                            promise: this.accepted.promise
                        }]
                    }
                }
                return null
            } else if (this.election.promises.length < this.election.majority.length) {
                return null
            } else {
                this.election.status = 'proposed'
                this.newGovernment(now, this.election.majority, {
                    majority: this.election.majority,
                    minority: this.election.minority
                }, this.promise)
            }
        } else {
            var parliament = this.government.majority.concat(this.government.minority)
// TODO The constituent must be both connected and synchronized, not just
// connected.
            var present = this.parliament.filter(function (id) {
                var peer = this._peers[id] || {}
                return id != this.id && peer.naturalized && peer.timeout == 0
            }.bind(this))
            var majoritySize = Math.ceil(parliament.length / 2)
            if (present.length < majoritySize) {
                return null
            }
            var majority = [ this.id ].concat(present).slice(0, majoritySize)
            var minority = this.parliament.filter(function (id) { return ! ~majority.indexOf(id) })
            this.election = {
                status: 'proposing',
                majority: majority,
                minority: minority,
                promises: [],
                accepts: []
            }
            return {
                type: 'consensus',
                islandId: this.islandId,
                governments: [ this.government.promise ],
                route: majority,
                messages: [{
                    type: 'propose',
                    // Do not increment here, it will be set by `_receivePromise`.
                    promise: Monotonic.increment(this.promise, 0)
                }]
            }
        }
    } else if (this.government.majority[0] == this.id && this.accepted && Monotonic.isBoundary(this.accepted.promise, 0)) {
        return {
            type: 'consensus',
            islandId: this.islandId,
            governments: [ this.government.promise, this.accepted.promise ],
            route: this.accepted.route,
            messages: [this._ping(now), {
                type: 'commit',
                promise: this.accepted.promise
            }]
        }
    } else if (this.immigrating.length && this.government.majority[0] == this.id) {
        var immigration = this.immigrating.shift()
        this.newGovernment(now, this.government.majority, {
            majority: this.government.majority,
            minority: this.government.minority,
            immigrate: {
                id: immigration.id,
                properties: immigration.properties,
                cookie: immigration.cookie
            }
        }, Monotonic.increment(this.promise, 0))
    } else if (this.ponged && this.id == this.government.majority[0]) {
        var reshape = this._impeach() || this._exile() || this._expand()
        if (reshape) {
            this.ponged = false
            this.newGovernment(now, reshape.quorum, reshape.government, Monotonic.increment(this.promise, 0))
        }
    }
    var proposal = this.proposals[0]
    var messages = [ this._ping(now) ]
    if (this.accepted != null) {
        messages.push({
            type: 'commit',
            promise: this.accepted.promise
        })
        if (proposal == null || !this._routeEqual(proposal.route, this.accepted.route)) {
            return {
                type: 'consensus',
                islandId: this.islandId,
                governments: [ this.government.promise ],
                route: this.accepted.route,
                messages: messages
            }
        }
    }
    if (proposal) {
        messages.push({
            type: 'accept',
            promise: proposal.promise,
            value: proposal.value
        })
        this.proposals.shift()
        return {
            type: 'consensus',
            islandId: this.islandId,
            governments: [ this.government.promise ],
            route: proposal.route.slice(),
            messages: messages
        }
    }
    if (this.keepAlive) {
        this.keepAlive = false
        return {
            type: 'consensus',
            islandId: this.islandId,
            governments: [ this.government.promise ],
            route: this.government.majority,
            messages: [ this._ping(now) ]
        }
    }
    return null
}

Legislator.prototype.consensus = function (now) {
    var pulse = null
    if (!this.pulsing) {
        pulse = this._consensus(now)
        this.pulsing = !! pulse
    }
    return pulse
}


Legislator.prototype.synchronize = function (now) {
    trace('synchronize', [ now ])
    var outbox = []
    for (var i = 0, I = this.constituency.length; i < I; i++) {
        var id = this.constituency[i]
        var peer = this.getPeer(id)
        var compare = Monotonic.compare(this.getPeer(id).decided, this.getPeer(this.id).decided)
        if ((peer.timeout != 0 || compare < 0) && !peer.skip && !this.synchronizing[id]) {
            var count = 20
            this.synchronizing[id] = true
            var pulse = {
                type: 'synchronize',
                islandId: this.islandId,
                governments: [ this.government.promise ],
                route: [ id ],
                messages: []
            }

            var maximum = peer.decided
            if (peer.pinged) {
                var round
                if (peer.decided == '0/0') {
                    var iterator = this.log.iterator()
                    for (;;) {
                        round = iterator.prev()
// TODO This will abend if the naturalization falls off the end end of the log.
// You need to check for gaps and missing naturalizations and then timeout the
// constituents that will never be connected.
                        assert(round, 'cannot find immigration')
                        if (Monotonic.isBoundary(round.promise, 0)) {
                            var immigrate = round.value.government.immigrate
                            if (immigrate && immigrate.id == id) {
                                maximum = round.promise
                                break
                            }
                        }
                    }
                } else {
// TODO Got a read property of null here.
                    round = this.log.find({ promise: maximum }).next
                }

                while (--count && round) {
                    pulse.messages.push({
                        type: 'enact',
                        promise: round.promise,
                        value: round.value
                    })
                    round = round.next
                }
            }

            pulse.messages.push(this._ping(now))

            outbox.push(pulse)
        }
    }
    return outbox
}

Legislator.prototype.receive = function (now, pulse, messages) {
    trace('_receive', [ now, pulse, messages ])
    assert(arguments.length == 3 && now != null)
    var responses = []
    for (var i = 0, I = messages.length; i < I; i++) {
        var message = messages[i]
        var type = message.type
        var method = '_receive' + type[0].toUpperCase() + type.substring(1)
        this[method].call(this, now, pulse, message, responses)
    }
    return responses
}

Legislator.prototype.collapse = function () {
    trace('collapse', [])
    this.collapsed = true
    this.election = null
    this.proposals.length = 0
    this.immigrating.length = 0
    for (var id in this._peers) {
        if (id != this.id) {
            delete this._peers[id]
        }
    }
    var parliament = this.government.majority.concat(this.government.minority)
    this.constituency = parliament.filter(function (id) {
        return this.id != id
    }.bind(this))
}

Legislator.prototype.sent = function (now, pulse, responses) {
    trace('sent', [ now, pulse, responses ])
    if (pulse.type == 'consensus') {
        this.pulsing = false
    }
    if (!~pulse.governments.indexOf(this.government.promise)) {
        return
    }
    var success = true
    pulse.route.forEach(function (id) {
        if (responses[id] == null) {
            success = false
        } else {
            this.receive(now, pulse, responses[id])
        }
    }, this)
    success = success && !pulse.failed
    if (success) {
        switch (pulse.type) {
        case 'synchronize':
            delete this.synchronizing[pulse.route[0]]
            this.scheduler.schedule(now + this.ping, pulse.route[0], { object: this, method: '_whenPing' }, pulse.route[0])
            break
        case 'consensus':
            this.scheduler.schedule(now + this.ping, this.id, { object: this, method: '_whenKeepAlive' })
{'fraction_non_alphanumeric': 0.11365949403896482, 'fraction_numerical': 0.0035620820005815646, 'mean_word_length': 1.7302768681155105, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '564868', 'n_tokens_mistral': 7841, 'n_tokens_neox': 7538, 'n_words': 1745}