protocol_node.go

package paxos

import "time"

type node struct {
	id             int
	clusterMembers []int

	// normally, they say nodes are either proposers or acceptors or learners
	// that doesn't mean we have different types of nodes necessarily, we can have a single type of nodes that can play all roles, they're more like hats, not strictly roles

	// ---- ACCEPTOR state ----
	promisedBallot   Ballot // highest ballot promised
	acceptedProposal *proposal

	// ---- PROPOSER state (optional) ----
	round       *proposerRound
	lastCounter int
	retryConfig retryConfig

	// ---- LEARNER state (...roughly) ----
	chosenValue *string

	// -- reusable output buffers --
	actions []nodeAction
	traces  []TraceEvent
}

// nodeEvents are triggers that happen to a node, coming from the outside world: proposalTriggered, messageReceived, retryTimerFired...
// the main takeaway is the signature of:

// func (n *node) Handle(event nodeEvent) nodeOutput
//
// that is, based on nodeEvent(s), we will (potentially) decide to take nodeAction(s) or record traceEvetn(s)

// example: based on a messageReceived that is a Prepare request from a proposer, the node might want to perform the action that will sendTo a promise to a proposer

// the caller of node.Handle(event) is responsible for interacting with the outside world and performing the node's decided next actions. I added this abstraction / decoupling layer because I needed to have both:
// 1- a normal in-memory, non-deteministic simulation run time
// 2- a deterministic scheduler for testing (testing this with timers and goroutiens is just ugly)
// but also it turned out to be more elegant!

// tldr: protocol_node.go is the protocol logic, with a state-machine-like shape
type nodeEvent interface {
	isNodeEvent()
}

// NOTE: the algorithm doesn't really specify what triggers proposals
// which signals that liveness is really achieved via external
// assumptions (e.g. there already exists a proposal...) unlike safety which _is_ guaranteed by the protocol under its failure modes (assuming it's correctly implemented)
type proposalTriggered struct {
	value string
}

func (proposalTriggered) isNodeEvent() {}

type messageReceived struct {
	from int
	msg  Message
}

func (messageReceived) isNodeEvent() {}

type retryTimerFired struct {
	ballot Ballot
}

func (retryTimerFired) isNodeEvent() {}

type nodeAction interface {
	isNodeAction()
}

type nodeOutput struct {
	actions []nodeAction
	traces  []TraceEvent
}

type sendMessage struct {
	to  int
	msg Message
}

func (sendMessage) isNodeAction() {}

type resetRetryTimer struct {
	ballot Ballot
	after  time.Duration
}

func (resetRetryTimer) isNodeAction() {}

type stopRetryTimer struct{}

func (stopRetryTimer) isNodeAction() {}

type TraceEvent interface {
	isTraceEvent()
}

type ValueLearned struct {
	Value string
}

func (ValueLearned) isTraceEvent() {}

type RoundStarted struct {
	Ballot Ballot
	Value  string
}

func (RoundStarted) isTraceEvent() {}

type RoundRetried struct {
	Ballot    Ballot
	Attempt   int
	NextAfter time.Duration
}

func (RoundRetried) isTraceEvent() {}

type RoundPreempted struct {
	Current Ballot
	Higher  Ballot
}

func (RoundPreempted) isTraceEvent() {}

type RoundAbandoned struct {
	Ballot Ballot
}

func (RoundAbandoned) isTraceEvent() {}

type PromiseQuorumReached struct {
	Ballot         Ballot
	Value          string
	CarriedForward bool
}

func (PromiseQuorumReached) isTraceEvent() {}

type proposerRound struct {
	promisers map[int]struct{}
	accepters map[int]struct{}

	highestAccepted *proposal

	ballot Ballot
	value  string

	retry retryState
}

type retryConfig struct {
	baseTimeout time.Duration
	maxDelay    time.Duration
	maxRetries  int
}

type retryState struct {
	currentTimeout time.Duration
	retryCount     int
}

func newNode(id int, clusterMembers []int, retryConfig retryConfig) *node {
	return &node{
		id:             id,
		clusterMembers: append([]int(nil), clusterMembers...),
		retryConfig:    retryConfig,
	}
}

func (n *node) Handle(event nodeEvent) nodeOutput {
	n.actions = n.actions[:0]
	n.traces = n.traces[:0]

	switch event := event.(type) {
	case proposalTriggered:
		n.startProposal(event.value, nil)
	case messageReceived:
		n.handleMessage(event.from, event.msg)
	case retryTimerFired:
		n.handleRetryTimer(event.ballot)
	}

	actions := append([]nodeAction(nil), n.actions...)
	traces := append([]TraceEvent(nil), n.traces...)
	n.actions = n.actions[:0]
	n.traces = n.traces[:0]
	return nodeOutput{
		actions: actions,
		traces:  traces,
	}
}

func (n *node) handleMessage(from int, msg Message) {
	switch msg := msg.(type) {
	case Prepare:
		n.handlePrepare(from, msg)
	case Promise:
		n.handlePromise(from, msg)
	case Accept:
		n.handleAccept(from, msg)
	case Accepted:
		n.handleAccepted(from, msg)
	case Reject:
		n.handleReject(msg)
	case Decided:
		n.handleDecided(msg)
	}
}

func (n *node) sendTo(to int, msg Message) {
	n.actions = append(n.actions, sendMessage{to: to, msg: msg})
}

func (n *node) sendToAll(msg Message) {
	for _, id := range n.clusterMembers {
		if id != n.id {
			n.sendTo(id, msg)
		}
	}
}

func (n *node) clusterQuorum() int {
	return len(n.clusterMembers)/2 + 1
}

func (n *node) resetRetryTimer(ballot Ballot, after time.Duration) {
	n.actions = append(n.actions, resetRetryTimer{ballot: ballot, after: after})
}

func (n *node) stopRetryTimer() {
	n.actions = append(n.actions, stopRetryTimer{})
}

func (n *node) recordTrace(event TraceEvent) {
	n.traces = append(n.traces, event)
}

func (n *node) startProposal(value string, above *Ballot) {
	newBallot := n.startProposalRound(value, above)
	n.resetRetryTimer(newBallot, n.round.retry.currentTimeout)
}

func (n *node) startProposalRound(value string, above *Ballot) Ballot {
	// we need the ballot to be higher than any we've seen, which leads to this dance:
	updatedCounter := n.lastCounter + 1
	if above != nil && above.Counter+1 > updatedCounter {
		updatedCounter = above.Counter + 1
	}
	if n.promisedBallot.Counter+1 > updatedCounter {
		updatedCounter = n.promisedBallot.Counter + 1
	}
	n.lastCounter = updatedCounter

	newBallot := Ballot{
		Counter:    updatedCounter,
		ProposerID: n.id,
	}

	n.round = &proposerRound{
		promisers: make(map[int]struct{}),
		accepters: make(map[int]struct{}),
		ballot:    newBallot,
		value:     value,
		retry:     retryState{currentTimeout: n.retryConfig.baseTimeout},
	}
	n.recordTrace(RoundStarted{Ballot: newBallot, Value: value})

	n.promisedBallot = newBallot
	n.round.promisers[n.id] = struct{}{}

	if n.acceptedProposal != nil {
		n.round.highestAccepted = n.acceptedProposal
	}

	n.sendToAll(Prepare{
		from:           n.id,
		proposedBallot: newBallot,
	})

	return newBallot
}

func (n *node) resetRetryBackoff() {
	state := n.round.retry
	cfg := n.retryConfig

	state.currentTimeout = cfg.baseTimeout
	state.retryCount = 0
	n.round.retry = state

	n.resetRetryTimer(n.round.ballot, state.currentTimeout)
}

func (n *node) handleRetryTimer(ballot Ballot) {
	if n.round == nil || ballot.cmp(n.round.ballot) != 0 {
		return
	}
	state := n.round.retry
	cfg := n.retryConfig

	if state.retryCount == cfg.maxRetries {
		n.recordTrace(RoundAbandoned{Ballot: n.round.ballot})
		n.round = nil
		n.stopRetryTimer()
		return
	}

	state.currentTimeout *= 2
	if state.currentTimeout > cfg.maxDelay {
		state.currentTimeout = cfg.maxDelay
	}

	state.retryCount++
	n.round.retry = state
	n.recordTrace(RoundRetried{
		Ballot:    n.round.ballot,
		Attempt:   state.retryCount,
		NextAfter: state.currentTimeout,
	})

	n.resetRetryTimer(n.round.ballot, state.currentTimeout)
	if len(n.round.promisers) < n.clusterQuorum() {
		n.sendToAll(Prepare{
			from:           n.id,
			proposedBallot: n.round.ballot,
		})
	} else {
		n.sendToAll(Accept{
			from:           n.id,
			proposedBallot: n.round.ballot,
			proposedValue:  n.round.value,
		})
	}
}

func (n *node) handlePrepare(from int, msg Prepare) {
	if msg.proposedBallot.cmp(n.promisedBallot) >= 0 {
		n.promisedBallot = msg.proposedBallot
		n.sendTo(from, Promise{
			from:             n.id,
			promisedBallot:   msg.proposedBallot,
			acceptedProposal: n.acceptedProposal,
		})
		return
	}

	n.sendTo(from, Reject{
		from:           n.id,
		promisedBallot: n.promisedBallot,
	})
}

func (n *node) handlePromise(from int, msg Promise) {
	round := n.round
	if round == nil {
		return
	}

	c := msg.promisedBallot.cmp(round.ballot)

	// c > 0: should arrive as a Reject, not a Promise
	// c < 0: stale, ignore
	if c != 0 {
		return
	}

	if _, ok := round.promisers[from]; ok {
		// be careful not to reset the timer for duplicate promises, that's fake liveness / progress
		return
	}
	n.resetRetryBackoff()
	round.promisers[from] = struct{}{}

	if msg.acceptedProposal != nil {
		incoming := msg.acceptedProposal
		if round.highestAccepted == nil || round.highestAccepted.ballot.cmp(incoming.ballot) < 0 {
			round.highestAccepted = incoming
		}
	}

	if len(round.promisers) == n.clusterQuorum() {
		// any acceptor already accepted a value? we must propose it then
		carriedForward := false
		if round.highestAccepted != nil {
			round.value = round.highestAccepted.value
			carriedForward = true
		}

		if round.ballot.cmp(n.promisedBallot) >= 0 {
			n.acceptedProposal = &proposal{
				ballot: round.ballot,
				value:  round.value,
			}
			round.accepters[n.id] = struct{}{}
		}

		n.recordTrace(PromiseQuorumReached{
			Ballot:         round.ballot,
			Value:          round.value,
			CarriedForward: carriedForward,
		})

		n.sendToAll(Accept{
			from:           n.id,
			proposedBallot: round.ballot,
			proposedValue:  round.value,
		})
	}
}

func (n *node) handleAccept(from int, msg Accept) {
	if msg.proposedBallot.cmp(n.promisedBallot) >= 0 {
		n.acceptedProposal = &proposal{
			ballot: msg.proposedBallot,
			value:  msg.proposedValue,
		}
		n.promisedBallot = msg.proposedBallot

		n.sendTo(from, Accepted{
			from:             n.id,
			promisedBallot:   n.promisedBallot,
			acceptedProposal: n.acceptedProposal,
		})
		return
	}

	n.sendTo(from, Reject{
		from:           n.id,
		promisedBallot: n.promisedBallot,
	})
}

func (n *node) handleAccepted(from int, msg Accepted) {
	round := n.round
	if round == nil {
		return
	}

	if msg.promisedBallot.cmp(round.ballot) != 0 {
		return
	}

	if _, ok := round.accepters[from]; ok {
		return
	}

	n.resetRetryBackoff()
	round.accepters[from] = struct{}{}

	if len(round.accepters) == n.clusterQuorum() {
		n.sendToAll(Decided{
			value: round.value,
		})
		n.chosenValue = &round.value
		n.recordTrace(ValueLearned{Value: round.value})

		n.round = nil
		n.stopRetryTimer()
	}
}

func (n *node) handleReject(msg Reject) {
	round := n.round
	if round == nil {
		return
	}

	if msg.promisedBallot.cmp(round.ballot) <= 0 {
		return
	}

	n.recordTrace(RoundPreempted{Current: round.ballot, Higher: msg.promisedBallot})
	n.startProposal(round.value, &msg.promisedBallot)
}

func (n *node) handleDecided(msg Decided) {
	if n.chosenValue != nil && *n.chosenValue != msg.value {
		panic("already chose a different value...")
	}

	if n.round != nil {
		n.round = nil
		n.stopRetryTimer()
	}

	if n.chosenValue == nil {
		n.chosenValue = &msg.value
		n.recordTrace(ValueLearned{Value: msg.value})
	}
}