program paxos_prepare;

import java.lang.System;

// inserted into from the calling program.
define(start, keys(0), {String});

/**** local_aru block ******/
/*
    ARU, or "all received upto", holds the value of the next globally ordered
    update we are ready to propose or write.  It should be initialized from
    stable storage: below, we initialize it to zero because we have none.

    When we successfully update our global history, we may increment the ARU.
    we do this via an aggregation to force a stratification boundary.
*/

define(local_aru, keys(0), {
  String, // host
  Integer // local ARU value
});

public
local_aru(Me, 0) :-
    paxos::self(Me),
    start(),
    notin global_history(Me, _, _, _);

/**************************/

/*
 * When a node wins leader election, multicast a prepare message to
 * the Paxos group containing the new view number and our local ARU.
 * The common wisdom says that the prepare phase of Paxos enforces
 * all the safety guarantees: an unreliable leader can at worst impact
 * liveness.  This is not true in this implementation, because the 
 * leader election module produces our View number.  The following 
 * safety constraints must hold over leader::leader(_, _, View):
 *  1. it must be unique to this host.
 *  2. it must be monotonically increasing.
 */

define(prepare, {
  String,  // Agent
  String,  // Master
  Integer, // View Number
  Integer  // ARU
});

public
prepare(@Him, Me, View, Aru) :-
    leader::leader(@Me, Leader, View),
    paxos::parliament(@Me, Him),
    local_aru(@Me, Aru),
    Leader == Me;

/**** datalist block *********************/
/*
 * When it receives a prepare message, an agent responds with a
 * prepare_ok message. This must contain:
 *      1. Any globally ordered updates with SeqNo > the ARU in the prepare message
 *      2. Any proposed values the agent has accepted, for previous views, 
 *         for which the same inequality holds.
 *
 * We eschew nested types, and combine the relevant attributes from the prepare
 * message with both lists above.
 */
define(datalist, keys(0, 1, 2), {
  String,   // Agent
  String,   // Master
  Integer,  // View Number
  Integer,  // ARU requested
  Integer,  // Sequence Number
  String,   // Update
  String    // Type (Bottom, Ordered, Proposed)
});

datalist(Agent, Master, View, Aru, SeqNo, Update, Type) :-
    global_history(Agent, SeqNo, _, Update),
    datalist(Agent, Master, View, Aru, -1, _, "Bottom"),
    SeqNo >= Aru,
    Type := "Ordered";

public
datalist(Agent, Master, View, Aru, SeqNo, Update, Type) :-
    accept(Agent, M, OldView, SeqNo, Update),
    datalist(Agent, Master, View, Aru, -1, _, "Bottom"),
    SeqNo >= Aru,
    Type := "Proposed";

// if SeqNo < Aru, we ignore the preoposal

// Rather than using explicit negation, we insert a dummy
// tuple into datalist to represent "nothing".
datalist(Agent, Master, View, Aru, -1, "none", "Bottom") :-
    prepare(Agent, Master, View, Aru),
    leader::last_installed(Agent, LastView),
    LastView == View;

define(datalist_length, keys(0, 1), {
  String,   // Agent
  Integer,  // ARU
  Integer   // count of datalist values
});

datalist_length(Agent, Aru, count<SeqNo>) :-
    datalist(Agent, _, _, Aru, SeqNo, _, _);
/*********************************************/



/******* prepare_oklist block ****************/
define(prepare_oklist, keys(0, 1, 2, 4, 5, 6), {
  String,   // Master
  Integer,  // View number
  Integer,  // Sequence number
  String,   // Update
  String,   // Type
  Integer,  // Datalist length
  String    // Agent
});

prepare_oklist(@Master, View, SeqNo, Update, Type, Len, Agent) :-
    datalist_length(@Agent, Aru, Len),
    datalist(@Agent, Master, View, Aru, SeqNo, Update, Type);


delete
prepare_oklist(Master, View, SeqNo, Update, Type, Len, Agent) :-
    prepare_oklist(Master, View, SeqNo, Update, Type, Len, Agent),
    global_history(Master, SeqNo, _, Update);

define(prepare_oklist_cnt, keys(0, 1, 2), {
  String,   // Master
  Integer,  // View number
  String,   // Agent
  Integer,  // Datalist length
  Integer   // Count
});

prepare_oklist_cnt(Master, View, Agent, Len, count<SeqNo>) :-
    prepare_oklist(Master, View, SeqNo, _, Type, Len, Agent);

define(prepare_ok_cnt, keys(0, 1), {
  String,   // Master
  Integer,  // View number
  Integer   // Agent Count
});

// only count agents who have sent prepare_ok messages towards
// the quorum if we're received the whole set of messages from
// that agent.
prepare_ok_cnt(Master, View, count<Agent>) :-
    prepare_oklist_cnt(Master, View, Agent, Cnt, Cnt2),
    Cnt == Cnt2;
/********************************************/


define(quorum, keys(0, 1), {
  String,   // Master
  Integer   // View
});

quorum(Master, View) :-
    paxos::priestCnt(Master, PCnt),
    leader::leader(Master, Leader, View),
    Master == Leader,
    prepare_ok_cnt(Master, View, RCnt),
    RCnt > (PCnt / 2);

define(accept, keys(0, 1), {
  String,   // Agent
  String,   // Master
  Integer,  // View number
  Integer,  // Sequence number
  String    // Update
});

define(global_history, keys(0, 1), {
  String,   // Master
  Integer,  // Sequence number
  String,   // Requestor
  String    // Update
});

global_history(Agent, SeqNo, Requestor, Update) :-
    prepare_oklist(Agent, View, SeqNo, Update, "Ordered", _, Requestor),
    notin global_history(Agent, SeqNo);