paxos.csp

// number of nodes
#define N 8; 

// entities
#define PROPOSERS 2; // please update for scenario 6
#define ACCEPTORS 3; // please update for scenario 6
#define LEARNERS 2;

// message types
enum {
	PASS,
	NACK,
	FAIL, 
	INACTIVE,
	ACCEPTOR_FAILED
};

// acceptor states
var acceptor_prepare_status = [PASS(ACCEPTORS)];
var acceptor_can_broadcast = [NACK(ACCEPTORS)];
var acceptor_promised_id = [-1(ACCEPTORS)];
var acceptor_accept_values = [-1(ACCEPTORS)];
var acceptor_accept_id = [-1(ACCEPTORS)];
var acceptor_status = [PASS(ACCEPTORS)];
var acceptor_number_of_accepts =[0(ACCEPTORS)];

// proposer states
var proposer_status = [PASS(PROPOSERS)];

// learner states
var learner_status = [PASS(LEARNERS)];

// global state for proposer
var global_accepted_value = -1;
var global_elected_leader = 0;
var global_terminate = false;
var global_id = 0;

// client state
var client_proposed_value = -1;
var client_read_value = -1;
var client_consensus = -1;
var client_can_propose = false;

// for validating quorums
var quorums = ACCEPTORS / 2 + 1;
var prepare_reads = 0;
var promise_received = 0;
var accept_reads = 0;
var accepted_received = 0;

// to indicate that the proposer should retry
var retry = false; 
var hasNack = false;
var highest_id = -1;
var last_accepted = -1;

// channels
channel reset[ACCEPTORS] 0;
channel prepare_channel[ACCEPTORS] 0;
channel promise_channel[ACCEPTORS] 0;
channel accepted_channel_learner[LEARNERS] 0;
channel accepted_channel_proposer 0;
channel accept_channel [ACCEPTORS] 0;

channel client_proposer 0;
channel learner_client 0;

// each proposer interleaves but only the elected leader can run
Proposer(state, i, canRecover) = 
	start -> 
	if (global_elected_leader == i) {
		// client may propose new values
		client_proposer?value {
			client_proposed_value = value;
			client_can_propose = false;
		} -> ProposerState(state, i, canRecover) [] ([client_proposed_value != -1] client_does_not_propose_value {client_can_propose = false} -> ProposerState(state, i, canRecover))
	} else {
		Stop
	};

// ProposerState tracks the current id of the proposer and increment it if required
ProposerState(state, curr, canRecover) = 
	if (global_id >= state) {
		// global_id is the last used global id 
		// so any requests afterwards will need to be higher
		ProposerState(global_id + 1, curr, canRecover) 
	} else {
		// Phase 1a. PREPARE
		SendPrepareMsg(state, curr);
		atomic { 
			if (prepare_reads < quorums) {
				atomic { PreparePhaseEnds() };
				atomic { (|||a:{0..ACCEPTORS-1}@(reset[a]!curr.FAIL -> Skip)) };
				ProposerState(state, curr, canRecover) 
		}};
		if (prepare_reads >= quorums) {
			atomic { PreparePhaseEnds() };
			// Phase 1b. PROMISE
			(|||y:{0..ACCEPTORS-1}@(promise_channel[y]?prev_promised_id.prev_acceptor_val.val{
				if (val == NACK) { global_terminate = true; global_id = prev_promised_id }
				if (val != INACTIVE && val != FAIL && val != ACCEPTOR_FAILED) { promise_received = promise_received + 1 }
			} -> Skip));
			atomic {
				if (promise_received < quorums) {
					atomic { 
						atomic { ResetPromiseReceived() };
						atomic { (|||a:{0..ACCEPTORS-1}@(reset[a]!curr.FAIL -> Skip)) };
						ProposerState(state, curr, canRecover) 
					}
			}};
			if (promise_received >= quorums) {
				atomic { ResetPromiseReceived() };
				// since we have achieved quorum of acceptors
				// next request must be larger than current state
				if (global_terminate) {
					atomic { (|||a:{0..ACCEPTORS-1}@(reset[a]!curr.FAIL -> Skip)) };
					ProposerState(state, curr, canRecover) // retry
				} else {
					atomic { promise_phase_ends { global_id = state } -> Skip };
					
					// if proposer fails BEFORE sending the first ACCEPT message
					// then there's no reason to proceed further because none of the acceptors will return a successful ACCEPTED message
					// hence we immediately reset the entire state
					(	// Phase 2a. ACCEPT      		 Phase 2b. ACCEPTED 
						(SendsAcceptMsg(curr, state) ||| ((|||{ACCEPTORS}@(accepted_channel_proposer?val.status {
									if (status == PASS) { client_consensus = val };
								} -> Skip)))); 
						atomic { end_of_broadcast -> Skip } ;
						if (global_terminate) {
							ElectNewLeader(state, curr, canRecover) [] Stop // if there are no more leaders to choose from, just stop
						} else if (retry) {
							atomic { (|||a:{0..ACCEPTORS-1}@(reset[a]!curr.FAIL -> Skip)) };
							// retry and elect ARE TWO DIFFERENT THINGS 
							// retry - proposer did not die
							// elect - proposer died
							ProposerState(state, curr, canRecover) 
						} else {
							Stop
						}	
					) 
				}
			}
		}
	};

// for electing a new leader, we check that the new leader is not the current, 
// and not any of the previously failed nodes
ElectNewLeader(state, curr, canRecover) = 
	if (curr == 0) {
		ElectNewLeaderRange(0, PROPOSERS - 1, state, curr, canRecover)
	};
	
	if (curr == PROPOSERS - 1) {
		ElectNewLeaderRange(0, PROPOSERS - 2, state, curr, canRecover)
	};
	
	if (0 < curr && curr < PROPOSERS - 1) {
		ElectNewLeaderRange(0, curr - 1, state, curr, canRecover) [] 
		ElectNewLeaderRange(curr + 1, PROPOSERS - 1, state, curr, canRecover) 
	};
	
ElectNewLeaderRange(low, high, state, curr, canRecover) = 
	[canRecover == true]  (
		[]i:{low..high}@([proposer_status[i] == PASS]
			(|||a:{0..ACCEPTORS-1}@(reset[a]!i.FAIL -> Skip));
			elect_new_leader {
				global_elected_leader = i;
				global_terminate = false;
				proposer_status[global_elected_leader] = PASS;
				accept_reads = 0; // reset reads
				client_can_propose = true;
			} -> (
				Proposer(acceptor_promised_id[i] + 1, i, canRecover) |||
				(recover { 
						global_id = state; 
						hasNack = false;
						global_elected_leader = curr;
						proposer_status[curr] = PASS;
					}-> atomic { 
						(|||a:{0..ACCEPTORS-1}@(reset[a]!curr.FAIL -> Skip))};
						Proposer(state, curr, canRecover)
					)
			)
		) 
	) [] [canRecover == false] (
		[]i:{low..high}@([proposer_status[i] == PASS]
			(|||a:{0..ACCEPTORS-1}@(reset[a]!i.FAIL -> Skip));
			elect_new_leader {
				global_elected_leader = i;
				global_terminate = false;
				proposer_status[global_elected_leader] = PASS;
				accept_reads = 0; // reset reads
				client_can_propose = true;
			} -> Proposer(acceptor_promised_id[i] + 1, i, canRecover) 
		) 
	);
	
SendPrepareMsg(state, curr) = 
	atomic { |||x:{0..ACCEPTORS-1}@(prepare_channel[x]!curr.state.PASS -> Skip [] prepare_channel[x]!curr.state.FAIL -> Skip) };
	
PreparePhaseEnds() = 
	prepare_ends { prepare_reads = 0; } -> Skip;

ResetPromiseReceived() = 
	reset_promise_count { promise_received = 0 } -> Skip;

// if no value is accepted previously, use the client_proposed_value
SendsAcceptMsg(curr, state) =  // an accept message 
	if (global_accepted_value == -1) { // no previously accepted value
		// by not making it atomic, this allows next step to be triggered once quorum is achieved.
		// if atomic, we can only move on once all acceptors have sent their accept.
		(|||z:{0..ACCEPTORS-1}@accept_channel[z]!proposer_status[curr].state.client_proposed_value.acceptor_status[z] -> Skip) 
		||| ((proposer_fails { 
			global_terminate = true; 
			proposer_status[curr] = FAIL;
		} -> Skip) [] (proposer_did_not_fail_when_sending_accept -> Skip))
	} else {
		(|||z:{0..ACCEPTORS-1}@accept_channel[z]!proposer_status[curr].state.global_accepted_value.acceptor_status[z] -> Skip) 
		||| ((proposer_fails { 
			global_terminate = true; 
			proposer_status[curr] = FAIL;
		} -> Skip) [] (proposer_did_not_fail_when_sending_accept -> Skip))
	}; 

Acceptor(i) = AcceptorState(i);
	
AcceptorState(i) = 
	(ReceivePrepare(i);
	((	
		SendPromise(i);
		// after receiving accept message, we need to broadcast the accepted to learners and proposers. TODO: proposers
		// valid only if the status of broadcast accepted is PASS
		// proposer fails while sending accept, each accept has to be broadcasted if received. need to send values into channel.
		((ReceivesAcceptMsg(i); BroadcastAccepted(i)) [] ResetAcceptor(i, true))
	)  [] ResetAcceptor(i, true)); 
	[global_terminate || retry] ResetAcceptor(i, false));
	
ReceivePrepare(i) = 
	prepare_channel[i]?curr.state.status { 
		if (status == FAIL) {
			// by default, it's true, so we set to fail
			acceptor_prepare_status[i] = FAIL
		} else {
			// retrieve the previous promised id (meaning subsequent id has to be larger than this)
			var prev_id = acceptor_accept_id[i];
			// increment # of prepare reads
			prepare_reads = prepare_reads + 1;
			// retrive the previous *accepted* value
			last_accepted = acceptor_accept_values[i];
			
			// if acceptor previously already accepted some value, 
			// and previous promised id is larger than the current highest_id we seen so far
			// then override the global_accepted_value
			// this global_accepted_value is the one that the proposer will use
			// client_proposed_value < global_accepted_value
			if (last_accepted != -1 && prev_id > highest_id) {
				highest_id = prev_id;
				global_accepted_value = last_accepted;
			}
			
			// we dont update the acceptor internal state during the prepare phase
			// but we still do the check
			if (acceptor_promised_id[i] >= state) { 
				hasNack = true; 
			} else {
				acceptor_promised_id[i] = state;
			}
		}
	} -> Skip;
	
ResetAcceptor(i, clear) = (
	reset[i]?curr.x {
		acceptor_can_broadcast[i] = NACK;
		acceptor_prepare_status[i] = PASS;
		if (clear == true) {
			hasNack = false;
			global_terminate = false;
			retry = false;
		}
	} -> AcceptorState(i)
) [] Stop;
	
ReceivesAcceptMsg(i) =
	accept_channel[i]?proposer_alive.state.av.status {
		if (status == PASS && proposer_alive == PASS) {
			accept_reads = accept_reads + 1;
			if (global_terminate == false) {
				var currentId = acceptor_promised_id[i];
				if (state >= currentId) {
					acceptor_accept_values[i] = av; // update accepted value
					acceptor_accept_id[i] = state; // update accepted value
					acceptor_can_broadcast[i] = PASS;
					acceptor_number_of_accepts[i] = acceptor_number_of_accepts[i] + 1;
				} else {
					retry = true; 
				}
			}
		}
	} -> Skip;
	
SendPromise(i) =
	if (acceptor_prepare_status[i] == FAIL || acceptor_status[i] == FAIL) {
		promise_channel[i]!acceptor_promised_id[i].acceptor_accept_values[i].INACTIVE-> Skip
	} else {
		([hasNack == true] promise_channel[i]!acceptor_promised_id[i].acceptor_accept_values[i].NACK -> Skip) []
		([hasNack == false] promise_channel[i]!acceptor_promised_id[i].acceptor_accept_values[i].PASS -> Skip) [] 
		(promise_channel[i]!acceptor_promised_id[i].acceptor_accept_values[i].FAIL -> Skip) [] 
		(promise_channel[i]!acceptor_promised_id[i].acceptor_accept_values[i].ACCEPTOR_FAILED { acceptor_status[i] = FAIL; acceptor_promised_id[i] = -1 } -> Skip) 
	};
	
BroadcastAccepted(i) = 
	((|||q:{0..LEARNERS-1}@accepted_channel_learner[q]!acceptor_accept_values[i].acceptor_can_broadcast[i]-> Skip) |||
	(accepted_channel_proposer!acceptor_accept_values[i].acceptor_can_broadcast[i]-> Skip));

Learner(i) =
	(|||{ACCEPTORS}@(accepted_channel_learner[i]?val.status {
		if (status == PASS) {
			client_consensus = val
		}
	} -> Skip)); 
	atomic {
		if (client_consensus == -1) {
			Learner(i)
		} else {
			(([accept_reads >= quorums] learner_client!i.client_consensus -> Stop) [] ([accept_reads >= quorums] learner_client!i.FAIL { learner_status[i] = FAIL } -> Stop) [] (reset_learner -> Learner(i)))
		}
	};


Client(value) =
		client_proposer!value -> (((|||{LEARNERS}@(learner_client?i.x {
		 	if (x != FAIL) {
			 	client_read_value = x
			}       // after proposing a value, you can interrupt and propose a new value
			        // that way when the next reset point happens you can set a new value
		} -> Skip)); atomic { finish { client_consensus = -1 } -> Stop }) []  [client_can_propose == true] (propose_new_value -> Client(value + 1)));

// entry point
var client_value = 123;
System = Client(client_value) || Proposer(global_id + 1, 0, true) 
					 || (|||x:{0..ACCEPTORS-1}@Acceptor(x)) 
					 || (|||y:{0..LEARNERS-1}@Learner(y));

// simplified model with no recover capabilities for proposer (to reduce number of states for verification)
SystemWithNoRecover = Client(client_value) || Proposer(global_id + 1, 0, false) 
					 || (|||x:{0..ACCEPTORS-1}@Acceptor(x)) 
					 || (|||y:{0..LEARNERS-1}@Learner(y));

// NOTE: For the verifications below, please use BFS.

// for verification
#define CONSENSUS_IS_ACHIEVED client_read_value == client_value;
#define CONSENSUS_IS_NOT_ACHIEVED client_read_value != client_value;
#define PROMISE_MAJORITY promise_received >= quorums;

// scenario 0
#assert System reaches CONSENSUS_IS_ACHIEVED; // there exists a trace where consensus is successfully achieved
#assert System |= !<>finish; // all traces does not finish - should be INVALID

// scenario 1
#define S1_CONSENSUS_IS_ACHIEVED client_read_value == client_value;
#define S1_CURRENT_ACCEPTOR_STATUS acceptor_status[0] == PASS && acceptor_status[1] == PASS && acceptor_status[2] == FAIL;
#define S1_CURRENT_ACCEPTOR_PROMISED_ID acceptor_promised_id[0] == 1 && acceptor_promised_id[1] == 1 && acceptor_promised_id[2] == -1;
#define S1_CURRENT_ACCEPTOR_CAN_BROADCAST_STATUS acceptor_can_broadcast[0] == PASS && acceptor_can_broadcast[1] == PASS && acceptor_can_broadcast[2] == NACK;

#define S1 S1_CURRENT_ACCEPTOR_STATUS && S1_CURRENT_ACCEPTOR_CAN_BROADCAST_STATUS && S1_CURRENT_ACCEPTOR_PROMISED_ID && S1_CONSENSUS_IS_ACHIEVED;
#assert System reaches S1;
#assert System |= !<>S1_CURRENT_ACCEPTOR_STATUS; // will never have acceptor failures - should be INVALID

// scenario 2
#define S2_CONSENSUS_IS_ACHIEVED client_read_value == client_value;
#define S2_LEARNER_STATUS learner_status[0] == FAIL && learner_status[1] == PASS;
#define S2_CURRENT_ACCEPTOR_STATUS acceptor_status[0] == PASS && acceptor_status[1] == PASS && acceptor_status[2] == PASS;
#define S2_CURRENT_ACCEPTOR_PROMISED_ID acceptor_promised_id[0] == 1 && acceptor_promised_id[1] == 1 && acceptor_promised_id[2] == 1;
#define S2_CURRENT_ACCEPTOR_CAN_BROADCAST_STATUS acceptor_can_broadcast[0] == PASS && acceptor_can_broadcast[1] == PASS && acceptor_can_broadcast[2] == PASS;

#define S2 S2_CURRENT_ACCEPTOR_STATUS && S2_CURRENT_ACCEPTOR_CAN_BROADCAST_STATUS && S2_CURRENT_ACCEPTOR_PROMISED_ID && S2_LEARNER_STATUS && S2_CONSENSUS_IS_ACHIEVED;
#assert System reaches S2;
#assert System |= !<>S2_LEARNER_STATUS; // will never have learner failures - should be INVALID

//// scenario 3
#define S3_NEW_ELECTED_LEADER global_elected_leader == 1;
#define S3_ELECTED_LEADER_NOT_FAIL proposer_status[0] == FAIL && proposer_status[1] == PASS; 
#define S3_NEW_CLIENT_VALUE client_proposed_value == 124;
#define S3 S3_NEW_ELECTED_LEADER && S3_ELECTED_LEADER_NOT_FAIL && S3_ELECTED_LEADER_NOT_FAIL && S3_NEW_CLIENT_VALUE;
#assert System reaches S3;
#assert System |= !<>proposer_fails; // will never have proposer_failures - should be INVALID
#assert SystemWithNoRecover |= [](proposer_fails -> G CONSENSUS_IS_NOT_ACHIEVED); // should be INVALID

// scenario 4
#define S4_CURRENT_ACCEPTOR_PROMISED_ID acceptor_promised_id[0] == 2 && acceptor_promised_id[1] == 2 && acceptor_promised_id[2] == 2 ;
#define S4_PROPOSER_STATUS proposer_status[0] == FAIL && proposer_status[1] == PASS;
#define S4 S4_PROPOSER_STATUS && S4_CURRENT_ACCEPTOR_PROMISED_ID;
#assert System reaches S4;
#assert System |= !<>recover; // will never recover - should be INVALID

// scenario 5
#define S5 acceptor_number_of_accepts[0] == 1; 
#assert SystemWithNoRecover |=  [](S5 -> G S5); // if accept_count == 1, doesnt mean it'll always remain as 1 therefore it should be INVALID

// scenario 6
#assert System |= !<>propose_new_value; // will never propose new value - should be INVALID

// scenario 7 
// if consensus is already achieved before the proposer fails, than reviving the proposer 
// or electing a new leader would not change the consensus.
#define S7_NEW_ELECTED_LEADER global_elected_leader == 1;
#define S7_ELECTED_LEADER_NOT_FAIL proposer_status[0] == FAIL && proposer_status[1] == PASS;
#define S7_NEW_CLIENT_VALUE client_proposed_value == 123;
#define ACCEPT_RECEIVED accept_reads == 1;
#define S7 S7_NEW_ELECTED_LEADER && S7_ELECTED_LEADER_NOT_FAIL && S7_NEW_CLIENT_VALUE && ACCEPT_RECEIVED;
#define S7_CONSENSUS_NOT_ACHIEVED client_read_value != client_value;
#assert SystemWithNoRecover |= [](S7 -> G S7_CONSENSUS_NOT_ACHIEVED);