Enable 2 round trip receipts (#999)

2020-04-01 10:38:46 +01:00 · 2020-04-01 10:38:46 +01:00 · 6f7b4dcbfc
--- a/src/consensus/pbft/libbyz/nv_info.cpp
+++ b/src/consensus/pbft/libbyz/nv_info.cpp
@ -891,7 +891,7 @@ Pre_prepare* NV_info::fetch_request(Seqno n, Digest& d, View& prev_view)
    // Null request
    Req_queue empty;
    size_t requests_in_batch;
-    pp = new Pre_prepare(v, n, empty, requests_in_batch);
+    pp = new Pre_prepare(v, n, empty, requests_in_batch, 0);
    pp->set_digest();
    d = pp->digest();
    prev_view = v;
--- a/src/consensus/pbft/libbyz/pre_prepare.cpp
+++ b/src/consensus/pbft/libbyz/pre_prepare.cpp
@ -19,6 +19,7 @@ Pre_prepare::Pre_prepare(
  Seqno s,
  Req_queue& reqs,
  size_t& requests_in_batch,
+  uint64_t nonce_,
  Prepared_cert* prepared_cert) :
  Message(
    Pre_prepare_tag,
@ -29,7 +30,8 @@ Pre_prepare::Pre_prepare(
      pbft::GlobalState::get_node().auth_size() + // Merkle root signature
      (pbft_max_signature_size + sizeof(uint64_t)) *
        pbft::GlobalState::get_node()
-          .num_of_replicas()) // signatures for the previous pre_prepare
+          .num_of_replicas()), // signatures for the previous pre_prepare
+  nonce(nonce_)
 {
  rep().view = v;
  rep().seqno = s;
@ -37,6 +39,12 @@ Pre_prepare::Pre_prepare(
  rep().contains_gov_req = false;
  rep().last_gov_req_updated = 0;

+  Digest dh;
+  Digest::Context context;
+  dh.update_last(context, (char*)&nonce_, sizeof(uint64_t));
+  dh.finalize(context);
+  rep().hashed_nonce = dh;
+
  START_CC(pp_digest_cycles);
  INCR_OP(pp_digest);

@ -258,6 +266,7 @@ bool Pre_prepare::calculate_digest(Digest& d)
      rep().replicated_state_merkle_root.size());
    d.update_last(context, (char*)&rep().contains_gov_req, sizeof(uint64_t));
    d.update_last(context, (char*)&rep().last_gov_req_updated, sizeof(Seqno));
+    d.update_last(context, (char*)&rep().hashed_nonce, sizeof(uint64_t));
    d.update_last(context, (char*)&rep().ctx, sizeof(rep().ctx));
    d.update_last(context, (char*)&rep().rset_size, sizeof(rep().rset_size));
    d.update_last(context, (char*)&rep().n_big_reqs, sizeof(rep().n_big_reqs));
--- a/src/consensus/pbft/libbyz/pre_prepare.h
+++ b/src/consensus/pbft/libbyz/pre_prepare.h
@ -35,6 +35,7 @@ struct Pre_prepare_rep : public Message_rep
  View view;
  Seqno seqno;
  std::array<uint8_t, MERKLE_ROOT_SIZE> replicated_state_merkle_root;
+  Digest hashed_nonce;
  uint64_t contains_gov_req; // should be a bool, but need to use 8 bytes to
                             // maintain alignment
  Seqno last_gov_req_updated;
@ -67,13 +68,14 @@ class Pre_prepare : public Message
  // Pre_prepare messages
  //
 public:
-  Pre_prepare(uint32_t msg_size = 0) : Message(msg_size) {}
+  Pre_prepare(uint32_t msg_size = 0) : Message(msg_size), nonce(0) {}

  Pre_prepare(
    View v,
    Seqno s,
    Req_queue& reqs,
    size_t& requests_in_batch,
+    uint64_t nonce,
    Prepared_cert* prepared_cert = nullptr);
  // Effects: Creates a new signed Pre_prepare message with view
  // number "v", sequence number "s", the requests in "reqs" (up to a
@ -228,12 +230,17 @@ public:
  bool is_signed();
  // Effects: checks if there is a signature over the pre_prepare message

+  uint64_t get_nonce() const;
+  // Effects: returns the unhashed nonce
+
  static bool convert(Message* m1, Pre_prepare*& m2);
  // Effects: If "m1" has the right size and tag, casts "m1" to a
  // "Pre_prepare" pointer, returns the pointer in "m2" and returns
  // true. Otherwise, it returns false.

 private:
+  uint64_t nonce;
+
  Pre_prepare_rep& rep() const;
  // Effects: Casts contents to a Pre_prepare_rep&

@ -328,3 +335,8 @@ inline bool Pre_prepare::did_exec_gov_req() const
 {
  return rep().contains_gov_req;
 }
+
+inline uint64_t Pre_prepare::get_nonce() const
+{
+  return nonce;
+}
--- a/src/consensus/pbft/libbyz/prepare.cpp
+++ b/src/consensus/pbft/libbyz/prepare.cpp
@ -12,17 +12,23 @@
 #include "replica.h"

 Prepare::Prepare(
-  View v, Seqno s, Digest& d, Principal* dst, bool is_signed, int id) :
+  View v,
+  Seqno s,
+  Digest& d,
+  uint64_t nonce_,
+  Principal* dst,
+  bool is_signed,
+  int id) :
  Message(
    Prepare_tag,
    sizeof(Prepare_rep)
 #ifndef USE_PKEY
-      + ((dst) ? MAC_size : pbft::GlobalState::get_node().auth_size()))
-{
+      + ((dst) ? MAC_size : pbft::GlobalState::get_node().auth_size())),
 #else
-      + ((dst) ? MAC_size : pbft_max_signature_size)
-{
+      + ((dst) ? MAC_size : pbft_max_signature_size)),
 #endif
+  nonce(nonce_)
+{
  rep().extra = (dst) ? 1 : 0;
  rep().view = v;
  rep().seqno = s;
@ -33,6 +39,12 @@ Prepare::Prepare(
    rep().id = pbft::GlobalState::get_node().id();
  }

+  Digest dh;
+  Digest::Context context;
+  dh.update_last(context, (char*)&nonce, sizeof(uint64_t));
+  dh.finalize(context);
+  rep().hashed_nonce = dh;
+
 #ifdef SIGN_BATCH
  rep().digest_sig_size = 0;
  rep().digest_padding.fill(0);
@ -43,11 +55,13 @@ Prepare::Prepare(
      uint32_t magic = 0xba5eba11;
      NodeId id;
      Digest d;
+      Digest n;

-      signature(Digest d_, NodeId id_) : d(d_), id(id_) {}
+      signature(Digest d_, NodeId id_, Digest nonce) : d(d_), id(id_), n(nonce)
+      {}
    };

-    signature s(d, rep().id);
+    signature s(d, rep().id, rep().hashed_nonce);

    rep().digest_sig_size = pbft::GlobalState::get_node().gen_signature(
      reinterpret_cast<char*>(&s), sizeof(s), rep().batch_digest_signature);
--- a/src/consensus/pbft/libbyz/prepare.h
+++ b/src/consensus/pbft/libbyz/prepare.h
@ -23,6 +23,7 @@ struct Prepare_rep : public Message_rep
  Seqno seqno;
  Digest digest;
  int id; // id of the replica that generated the message.
+  Digest hashed_nonce;
 #ifdef SIGN_BATCH
  size_t digest_sig_size;
  PbftSignature batch_digest_signature;
@ -50,12 +51,13 @@ class Prepare : public Message
  // Prepare messages
  //
 public:
-  Prepare(uint32_t msg_size = 0) : Message(msg_size) {}
+  Prepare(uint32_t msg_size = 0) : Message(msg_size), nonce(0) {}

  Prepare(
    View v,
    Seqno s,
    Digest& d,
+    uint64_t nonce,
    Principal* dst = 0,
    bool is_signed = false,
    int id = -1);
@ -99,12 +101,17 @@ public:
  bool pre_verify();
  // Effects: Performs preliminary verification checks

+  uint64_t get_nonce() const;
+  // Effects: returns the unhashed nonce
+
  static bool convert(Message* m1, Prepare*& m2);
  // Effects: If "m1" has the right size and tag, casts "m1" to a
  // "Prepare" pointer, returns the pointer in "m2" and returns
  // true. Otherwise, it returns false.

 private:
+  uint64_t nonce;
+
  Prepare_rep& rep() const;
  // Effects: Casts contents to a Prepare_rep&
 };
@ -152,3 +159,8 @@ inline bool Prepare::match(const Prepare* p) const
  PBFT_ASSERT(view() == p->view() && seqno() == p->seqno(), "Invalid argument");
  return digest() == p->digest();
 }
+
+inline uint64_t Prepare::get_nonce() const
+{
+  return nonce;
+}
--- a/src/consensus/pbft/libbyz/receive_message_base.h
+++ b/src/consensus/pbft/libbyz/receive_message_base.h
@ -48,6 +48,6 @@ public:
  virtual void playback_pre_prepare(ccf::Store::Tx& tx) = 0;
  virtual void playback_request(ccf::Store::Tx& tx) = 0;
  virtual char* create_response_message(
-    int client_id, Request_id rid, uint32_t size) = 0;
+    int client_id, Request_id rid, uint32_t size, uint64_t nonce) = 0;
  virtual bool IsExecutionPending() = 0;
 };
--- a/src/consensus/pbft/libbyz/rep_info.cpp
+++ b/src/consensus/pbft/libbyz/rep_info.cpp
@ -34,10 +34,10 @@ void Rep_info::count_request()
 }

 char* Rep_info::new_reply(
-  int pid, Request_id rid, Seqno n, uint32_t message_size)
+  int pid, Request_id rid, Seqno n, uint64_t nonce, uint32_t message_size)
 {
  message_size += sizeof(Reply_rep) + MAC_size;
-  auto r = std::make_unique<Reply>(0, rid, n, 0, message_size);
+  auto r = std::make_unique<Reply>(0, rid, n, nonce, 0, message_size);
  PBFT_ASSERT(r != nullptr, "Out of memory");
  r->set_size(message_size);
  r->trim();
--- a/src/consensus/pbft/libbyz/rep_info.h
+++ b/src/consensus/pbft/libbyz/rep_info.h
@ -42,7 +42,8 @@ public:
  // how many individual requests have been processes since the
  // replica was initialized. Should be called once for each request.

-  char* new_reply(int pid, Request_id rid, Seqno n, uint32_t message_size);
+  char* new_reply(
+    int pid, Request_id rid, Seqno n, uint64_t nonce, uint32_t message_size);
  // Effects: Allocates a new reply for request rid from
  // principal pid executed at sequence number n and returns a buffer
  // to store the reply to the command. The buffer can store up to
--- a/src/consensus/pbft/libbyz/replica.cpp
+++ b/src/consensus/pbft/libbyz/replica.cpp
@ -73,6 +73,7 @@ Replica::Replica(
 #endif
  rep_cb(nullptr),
  global_commit_cb(nullptr),
+  entropy(tls::create_entropy()),
  state(
    this,
    mem,
@ -477,7 +478,7 @@ void Replica::playback_request(ccf::Store::Tx& tx)
  vec_exec_cmds[0] = std::move(execute_tentative_request(
    *req, playback_max_local_commit_value, true, &tx, true));

-  exec_command(vec_exec_cmds, playback_byz_info, 1);
+  exec_command(vec_exec_cmds, playback_byz_info, 1, 0);
  did_exec_gov_req = did_exec_gov_req || playback_byz_info.did_exec_gov_req;
  brt.add_request(req.release());
 }
@ -515,6 +516,7 @@ void Replica::populate_certificates(Pre_prepare* pp, bool add_mine)
        prev_pp->view(),
        prev_pp->seqno(),
        prev_pp->digest(),
+        prev_pp->get_nonce(),
        nullptr,
        prev_pp->is_signed(),
        p_id);
@ -527,6 +529,7 @@ void Replica::populate_certificates(Pre_prepare* pp, bool add_mine)
        prev_pp->view(),
        prev_pp->seqno(),
        prev_pp->digest(),
+        prev_pp->get_nonce(),
        nullptr,
        prev_pp->is_signed());
      prev_prepared_cert.add_mine(p);
@ -924,6 +927,7 @@ void Replica::send_pre_prepare(bool do_not_wait_for_batch_size)
    LOG_TRACE << "creating pre prepare with seqno: " << next_pp_seqno
              << std::endl;
    auto ctx = std::make_unique<ExecTentativeCbCtx>();
+    ctx->nonce = entropy->random64();

    Prepared_cert* ps = nullptr;
    if (next_pp_seqno > congestion_window)
@ -931,7 +935,7 @@ void Replica::send_pre_prepare(bool do_not_wait_for_batch_size)
      ps = &plog.fetch(next_pp_seqno - congestion_window);
    }
    Pre_prepare* pp = new Pre_prepare(
-      view(), next_pp_seqno, rqueue, ctx->requests_in_batch, ps);
+      view(), next_pp_seqno, rqueue, ctx->requests_in_batch, ctx->nonce, ps);

    auto fn = [](
                Pre_prepare* pp,
@ -1145,7 +1149,12 @@ void Replica::send_prepare(Seqno seqno, std::optional<ByzInfo> byz_info)
        }

        Prepare* p = new Prepare(
-          self->v, pp->seqno(), pp->digest(), nullptr, pp->is_signed());
+          self->v,
+          pp->seqno(),
+          pp->digest(),
+          msg->nonce,
+          nullptr,
+          pp->is_signed());
        int send_node_id =
          (msg->send_only_to_self ? self->node_id : All_replicas);
        self->send(p, send_node_id);
@ -1168,6 +1177,7 @@ void Replica::send_prepare(Seqno seqno, std::optional<ByzInfo> byz_info)
      msg->seqno = seqno;
      msg->send_only_to_self = send_only_to_self;
      msg->orig_byzinfo = byz_info;
+      msg->nonce = entropy->random64();
      if (byz_info.has_value())
      {
        msg->info = byz_info.value();
@ -1497,9 +1507,10 @@ int Replica::my_id() const
 }

 char* Replica::create_response_message(
-  int client_id, Request_id request_id, uint32_t size)
+  int client_id, Request_id request_id, uint32_t size, uint64_t nonce)
 {
-  return replies.new_reply(client_id, request_id, last_tentative_execute, size);
+  return replies.new_reply(
+    client_id, request_id, last_tentative_execute, nonce, size);
 }

 void Replica::handle(Status* m)
@ -2069,7 +2080,7 @@ void Replica::process_new_view(Seqno min, Digest d, Seqno max, Seqno ms)
    {
      ByzInfo info;
      pc.add_mine(pp);
-      if (execute_tentative(pp, info))
+      if (execute_tentative(pp, info, pp->get_nonce()))
      {
        if (ledger_writer && req_in_pp > 0)
        {
@ -2081,7 +2092,9 @@ void Replica::process_new_view(Seqno min, Digest d, Seqno max, Seqno ms)
    {
      ByzInfo info;
      pc.add_old(pp);
-      if (execute_tentative(pp, info))
+      uint64_t nonce = entropy->random64();
+
+      if (execute_tentative(pp, info, nonce))
      {
        if (ledger_writer && req_in_pp > 0)
        {
@ -2089,7 +2102,7 @@ void Replica::process_new_view(Seqno min, Digest d, Seqno max, Seqno ms)
        }
      }

-      Prepare* p = new Prepare(v, i, d, nullptr, pp->is_signed());
+      Prepare* p = new Prepare(v, i, d, nonce, nullptr, pp->is_signed());
      pc.add_mine(p);
      send(p, All_replicas);
    }
@ -2375,7 +2388,7 @@ bool Replica::create_execute_commands(
  return false;
 }

-bool Replica::execute_tentative(Pre_prepare* pp, ByzInfo& info)
+bool Replica::execute_tentative(Pre_prepare* pp, ByzInfo& info, uint64_t nonce)
 {
  LOG_DEBUG_FMT(
    "in execute tentative for seqno {} and last_tentnative_execute {}",
@ -2386,7 +2399,7 @@ bool Replica::execute_tentative(Pre_prepare* pp, ByzInfo& info)
  if (create_execute_commands(
        pp, info.max_local_commit_value, vec_exec_cmds, num_requests))
  {
-    exec_command(vec_exec_cmds, info, num_requests);
+    exec_command(vec_exec_cmds, info, num_requests, nonce);
    return true;
  }
  return false;
@ -2408,6 +2421,7 @@ bool Replica::execute_tentative(
  if (create_execute_commands(
        pp, ctx->info.max_local_commit_value, vec_exec_cmds, num_requests))
  {
+    uint64_t nonce = ctx->nonce;
    ByzInfo& info = ctx->info;
    if (cb != nullptr)
    {
@ -2428,7 +2442,7 @@ bool Replica::execute_tentative(
      }
    }

-    exec_command(vec_exec_cmds, info, num_requests);
+    exec_command(vec_exec_cmds, info, num_requests, nonce);
    if (!node_info.general_info.support_threading)
    {
      cb(pp, this, std::move(ctx));
@ -2480,7 +2494,7 @@ void Replica::execute_committed(bool was_f_0)
        if (last_executed + 1 > last_tentative_execute)
        {
          ByzInfo info;
-          auto executed_ok = execute_tentative(pp, info);
+          auto executed_ok = execute_tentative(pp, info, pp->get_nonce());
          PBFT_ASSERT(
            executed_ok,
            "tentative execution while executing committed failed");
@ -3134,8 +3148,9 @@ void Replica::send_null()
        ps = &plog.fetch(next_pp_seqno - 1);
      }

-      Pre_prepare* pp =
-        new Pre_prepare(view(), next_pp_seqno, empty, requests_in_batch, ps);
+      uint64_t nonce = entropy->random64();
+      Pre_prepare* pp = new Pre_prepare(
+        view(), next_pp_seqno, empty, requests_in_batch, nonce, ps);
      pp->set_digest();
      send(pp, All_replicas);
      plog.fetch(next_pp_seqno).add_mine(pp);
--- a/src/consensus/pbft/libbyz/replica.h
+++ b/src/consensus/pbft/libbyz/replica.h
@ -137,7 +137,8 @@ public:
  void send(Message* m, int i);
  Seqno get_last_executed() const;
  int my_id() const;
-  char* create_response_message(int client_id, Request_id rid, uint32_t size);
+  char* create_response_message(
+    int client_id, Request_id rid, uint32_t size, uint64_t nonce);

  // variables used to keep track of versions so that we can tell the kv to
  // rollback
@ -312,6 +313,7 @@ private:
    Seqno seqno;
    bool send_only_to_self = false;
    std::optional<ByzInfo> orig_byzinfo;
+    uint64_t nonce;
  };

  struct ExecuteTentativeCbMsg
@ -335,7 +337,7 @@ private:
    std::array<std::unique_ptr<ExecCommandMsg>, Max_requests_in_batch>& cmds,
    uint32_t& num_requests);

-  bool execute_tentative(Pre_prepare* pp, ByzInfo& info);
+  bool execute_tentative(Pre_prepare* pp, ByzInfo& info, uint64_t nonce);

  bool execute_tentative(
    Pre_prepare* pp,
@ -582,6 +584,8 @@ private:
  // primary it will buffer messages until the other nodes have successfully
  // opened their networks

+  std::shared_ptr<tls::Entropy> entropy;
+
 #ifdef DEBUG_SLOW
  std::unique_ptr<ITimer> debug_slow_timer; // Used to dump state when requests
                                            // take too long to execute
--- a/src/consensus/pbft/libbyz/reply.cpp
+++ b/src/consensus/pbft/libbyz/reply.cpp
@ -12,12 +12,18 @@
 #include "statistics.h"

 Reply::Reply(
-  View view, Request_id req, Seqno n, int replica, uint32_t reply_size) :
+  View view,
+  Request_id req,
+  Seqno n,
+  uint64_t nonce,
+  int replica,
+  uint32_t reply_size) :
  Message(Reply_tag, sizeof(Reply_rep) + reply_size + MAC_size)
 {
  rep().v = view;
  rep().rid = req;
  rep().n = n;
+  rep().nonce = nonce;
  rep().replica = replica;
  rep().reply_size = 0;
  set_size(sizeof(Reply_rep) + reply_size + MAC_size);
@ -29,6 +35,7 @@ Reply::Reply(
  View view,
  Request_id req,
  Seqno n,
+  uint64_t nonce,
  int replica,
  Principal* p,
  bool tentative) :
@ -46,6 +53,7 @@ Reply::Reply(
  rep().v = view;
  rep().rid = req;
  rep().n = n;
+  rep().nonce = nonce;
  rep().replica = replica;
  rep().reply_size = -1;

--- a/src/consensus/pbft/libbyz/reply.h
+++ b/src/consensus/pbft/libbyz/reply.h
@ -22,6 +22,8 @@ struct Reply_rep : public Message_rep
  View v; // current view
  Request_id rid; // unique request identifier
  Seqno n; // sequence number when request was executed
+  uint64_t nonce; // plain text pre-prepare or prepare nonce that will be sent
+                  // to the client
  int replica; // id of replica sending the reply
  int reply_size; // if negative, reply is not full.
  // Followed by a reply that is "reply_size" bytes long and
@ -47,7 +49,13 @@ public:

  Reply(Reply_rep* r);

-  Reply(View view, Request_id req, Seqno n, int replica, uint32_t reply_size);
+  Reply(
+    View view,
+    Request_id req,
+    Seqno n,
+    uint64_t nonce,
+    int replica,
+    uint32_t reply_size);
  // Effects: Creates a new (full) Reply message with an empty reply and no
  // authentication. The method store_reply and authenticate should
  // be used to finish message construction.
@ -75,6 +83,7 @@ public:
    View view,
    Request_id req,
    Seqno n,
+    uint64_t nonce,
    int replica,
    Principal* p,
    bool tentative);
--- a/src/consensus/pbft/libbyz/test/replica_test.cpp
+++ b/src/consensus/pbft/libbyz/test/replica_test.cpp
@ -0,0 +1,449 @@
+// Copyright (c) Microsoft Corporation.
+// Copyright (c) 1999 Miguel Castro, Barbara Liskov.
+// Copyright (c) 2000, 2001 Miguel Castro, Rodrigo Rodrigues, Barbara Liskov.
+// Licensed under the MIT license.
+
+#include <CLI11/CLI11.hpp>
+#include <iostream>
+#include <signal.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <sys/param.h>
+#include <unistd.h>
+
+extern "C"
+{
+#include <evercrypt/EverCrypt_AutoConfig2.h>
+}
+
+#include "big_req_table.h"
+#include "client_proxy.h"
+#include "consensus/pbft/pbft_tables.h"
+#include "ds/files.h"
+#include "ds/thread_messaging.h"
+#include "host/ledger.h"
+#include "itimer.h"
+#include "libbyz.h"
+#include "network_impl.h"
+#include "nodeinfo.h"
+#include "pbft_assert.h"
+#include "replica.h"
+#include "stacktrace_utils.h"
+#include "statistics.h"
+#include "test_message.h"
+#include "timer.h"
+
+using std::cerr;
+
+static const int Simple_size = 4096;
+
+enclave::ThreadMessaging enclave::ThreadMessaging::thread_messaging;
+std::atomic<uint16_t> enclave::ThreadMessaging::thread_count = 0;
+
+static Timer t;
+static ITimer* test_timer;
+static ITimer* delay_test_timer;
+int random_delay_order = 100000;
+int test_timer_order = 1000;
+bool with_timeouts = false;
+bool write_to_ledger = false;
+bool have_executed_request = false;
+const int max_num_principals = 1000;
+int broken_requests[max_num_principals];
+
+static void dump_profile(int sig)
+{
+  unsigned short buf;
+  profil(&buf, 0, 0, 0);
+
+  LOG_INFO << "Printing stats" << std::endl;
+  stats.print_stats();
+
+  exit(0);
+}
+
+void introduce_random_delay()
+{
+  if (have_executed_request)
+  {
+    auto random_delay = random_delay_order * (lrand48() % 100);
+    LOG_INFO << "Sleeping for microseconds: " << random_delay << std::endl;
+
+    usleep(random_delay);
+    have_executed_request = false;
+  }
+  else
+  {
+    LOG_INFO << "Skipping Sleeping " << std::endl;
+  }
+}
+
+void test_timer_handler(void* owner)
+{
+  introduce_random_delay();
+  test_timer->restart();
+}
+
+void delayed_start_delay_time(void* owner)
+{
+  if ((pbft::GlobalState::get_replica().id() % 2) == 0) // half the nodes
+                                                        // including the primary
+  {
+    auto delay = 10 * 1000 * 1000; // sleep for 10 seconds
+    LOG_INFO << "Sleeping for " << (delay / (1000 * 1000))
+             << " seconds to force view change" << std::endl;
+    usleep(delay);
+  }
+
+  auto timeout_time = test_timer_order * ((lrand48() % 100) + 1);
+  LOG_INFO << "Init timer with milliseconds " << timeout_time << std::endl;
+  test_timer = new ITimer(timeout_time, test_timer_handler, nullptr);
+  test_timer->start();
+}
+
+void start_delay_timer()
+{
+  auto delay = 5 * 1000; // sleep in 5 seconds
+  if ((pbft::GlobalState::get_replica().id() % 2) == 0) // half the nodes
+                                                        // including the primary
+  {
+    delay += 10 * 1000; // make sure that all the replicas do not sleep when
+                        // enforcing the first view change
+  }
+
+  delay_test_timer = new ITimer(delay, delayed_start_delay_time, nullptr);
+  delay_test_timer->start();
+}
+
+static std::unique_ptr<ClientProxy<uint64_t, void>> client_proxy;
+static std::unique_ptr<ITimer> send_req_timer;
+static const size_t client_proxy_req_size = 8;
+static size_t reply_count = 0;
+static size_t request_count = 0;
+void setup_client_proxy()
+{
+  LOG_INFO << "Setting up client proxy " << std::endl;
+  client_proxy.reset(
+    new ClientProxy<uint64_t, void>(pbft::GlobalState::get_replica()));
+
+  auto cb = [](Reply* m, void* ctx) {
+    auto cp = (ClientProxy<uint64_t, void>*)ctx;
+    cp->recv_reply(m);
+  };
+  pbft::GlobalState::get_replica().register_reply_handler(
+    cb, client_proxy.get());
+
+  auto req_timer_cb = [](void* ctx) {
+    auto cp = (ClientProxy<uint64_t, void>*)ctx;
+
+    static const uint32_t max_pending_requests = 7;
+    while (request_count - reply_count < max_pending_requests)
+    {
+      uint8_t request_buffer[8];
+      auto request = new (request_buffer) test_req;
+      Time t = ITimer::current_time();
+
+      request->option = 0;
+      for (size_t j = 0; j < request->get_array_size(client_proxy_req_size);
+           j++)
+      {
+        memcpy(&request->get_counter_array()[j], &t, sizeof(int64_t));
+      }
+
+      auto rep_cb = [](
+                      void* owner,
+                      uint64_t caller_rid,
+                      int status,
+                      uint8_t* reply,
+                      size_t len) {
+        reply_count++;
+
+        if (reply_count % 100 == 0)
+        {
+          LOG_INFO << " Reply count " << reply_count << std::endl;
+        }
+        return true;
+      };
+
+      bool ret = cp->send_request(
+        t, request_buffer, sizeof(request_buffer), rep_cb, client_proxy.get());
+      if (ret)
+      {
+        request_count++;
+      }
+      else
+      {
+        break;
+      }
+    }
+
+    send_req_timer->restart();
+  };
+
+  send_req_timer.reset(new ITimer(100, req_timer_cb, client_proxy.get()));
+  send_req_timer->start();
+}
+
+static char* service_mem = 0;
+static IMessageReceiveBase* message_receive_base;
+
+ExecCommand exec_command =
+  [](
+    std::array<std::unique_ptr<ExecCommandMsg>, Max_requests_in_batch>& msgs,
+    ByzInfo& info,
+    uint32_t num_requests,
+    uint64_t nonce) {
+    for (uint32_t i = 0; i < num_requests; ++i)
+    {
+      std::unique_ptr<ExecCommandMsg>& msg = msgs[i];
+      Byz_req* inb = &msg->inb;
+      Byz_rep& outb = msg->outb;
+      int client = msg->client;
+      Request_id rid = msg->rid;
+      uint8_t* req_start = msg->req_start;
+      size_t req_size = msg->req_size;
+      Seqno total_requests_executed = msg->total_requests_executed;
+      ccf::Store::Tx* tx = msg->tx;
+
+      outb.contents =
+        message_receive_base->create_response_message(client, rid, 8, nonce);
+
+      Long& counter = *(Long*)service_mem;
+      Long* client_counter_arrays = (Long*)service_mem + sizeof(Long);
+      auto client_counter = client_counter_arrays[client];
+
+      Byz_modify(&counter, sizeof(counter));
+      counter++;
+      have_executed_request = true;
+
+      info.replicated_state_merkle_root.fill(0);
+      ((Long*)(info.replicated_state_merkle_root.data()))[0] = counter;
+      info.ctx = counter;
+
+      if (total_requests_executed != counter)
+      {
+        LOG_FATAL << "total requests executed: " << total_requests_executed
+                  << " not equal to exec command counter: " << counter << "\n";
+        throw std::logic_error(
+          "Total requests executed not equal to exec command counter");
+      }
+
+      if (total_requests_executed % 100 == 0)
+      {
+        LOG_INFO << "total requests executed " << total_requests_executed
+                 << "\n";
+      }
+
+      auto request = new (inb->contents) test_req;
+
+      for (size_t j = 0; j < request->get_array_size(inb->size); j++)
+      {
+        uint64_t request_array_counter;
+        memcpy(
+          &request_array_counter,
+          &request->get_counter_array()[j],
+          sizeof(uint64_t));
+
+        if (client_counter != request_array_counter && !broken_requests[client])
+        {
+          broken_requests[client] = 1;
+          LOG_INFO << "client: " << client
+                   << " broken state: " << broken_requests[client] << std::endl;
+          LOG_INFO << "client: " << client << std::endl;
+          LOG_INFO << "client counter: " << client_counter
+                   << " is smaller than request counter: "
+                   << request_array_counter << "\n";
+        }
+        else if (
+          client_counter == request_array_counter && broken_requests[client])
+        {
+          LOG_INFO << "client: " << client << std::endl;
+          broken_requests[client] = 0;
+          LOG_INFO << "client: " << client
+                   << " broken state: " << broken_requests[client] << std::endl;
+          LOG_INFO << "Fixed c counter: " << client_counter
+                   << " is NOT smaller than request counter: "
+                   << request_array_counter << "\n";
+        }
+      }
+
+      Byz_modify(&client_counter_arrays[client], sizeof(Long));
+      client_counter_arrays[client] = ++client_counter;
+
+      // A simple service.
+      if (request->option == 1)
+      {
+        PBFT_ASSERT(inb->size == 8, "Invalid request");
+        Byz_modify(outb.contents, Simple_size);
+        bzero(outb.contents, Simple_size);
+        outb.size = Simple_size;
+        return 0;
+      }
+
+      PBFT_ASSERT(
+        (request->option == 2 && inb->size == Simple_size) ||
+          (request->option == 0 && inb->size == 8),
+        "Invalid request");
+      Byz_modify(outb.contents, 8);
+      *((long long*)(outb.contents)) = 0;
+      outb.size = 8;
+      msg->cb(*msg.get(), info);
+    }
+    return 0;
+  };
+
+int main(int argc, char** argv)
+{
+  CLI::App app{"Run Replica Test"};
+
+  // run tests
+  short port = 0;
+  app.add_option("--port", port, "Port", true);
+
+  bool print_to_stdout = false;
+  app.add_flag("--stdout", print_to_stdout);
+
+  std::string transport_layer = "UDP";
+  app.add_option(
+    "--transport", transport_layer, "Transport layer [UDP || UDP_MT]");
+
+  app.add_option(
+    "--timer-order",
+    test_timer_order,
+    "Order of magnitued for the test timer timeout intervals");
+  app.add_option(
+    "--delay-order", random_delay_order, "Order of mangitude of random delay");
+
+  std::string config_file = "config.json";
+  app.add_option("--config", config_file, "General config info", true)
+    ->check(CLI::ExistingFile);
+
+  NodeId id;
+  app.add_option("--id", id, "Nodes id", true);
+
+  std::string privk_file;
+  app.add_option("--privk_file", privk_file, "Private key file", true)
+    ->check(CLI::ExistingFile);
+
+  app.add_flag("--with-delays", with_timeouts, "Insert delays");
+
+  app.add_flag("--ledger", write_to_ledger, "Should write to ledger");
+
+  bool test_client_proxy = false;
+  app.add_flag("--test-client-proxy", test_client_proxy, "Test client proxy");
+
+  CLI11_PARSE(app, argc, argv);
+
+  if (!print_to_stdout)
+  {
+    logger::Init(std::to_string(port).c_str());
+  }
+
+  Log_allocator::should_use_malloc(true);
+
+  GeneralInfo general_info = files::slurp_json(config_file);
+
+  general_info.max_requests_between_signatures = 10;
+
+  // as to not add double escapes on newline when slurping from file
+  PrivateKey privk_j = files::slurp_json(privk_file);
+  NodeInfo node_info;
+  tls::KeyPairPtr kp = tls::make_key_pair(privk_j.privk);
+  auto node_cert = kp->self_sign("CN=CCF node");
+
+  for (auto& pi : general_info.principal_info)
+  {
+    pi.cert = node_cert;
+  }
+
+  for (auto& pi : general_info.principal_info)
+  {
+    if (pi.id == id)
+    {
+      node_info = {pi, privk_j.privk, general_info};
+      break;
+    }
+  }
+
+  LOG_INFO << "Printing command line arguments" << std::endl;
+  std::stringstream cmd_line;
+  for (int i = 0; i < argc; ++i)
+  {
+    cmd_line << argv[i] << " ";
+  }
+  LOG_INFO << cmd_line.str() << std::endl;
+
+  LOG_INFO << "Starting replica main" << std::endl;
+
+  EverCrypt_AutoConfig2_init();
+
+  // signal handler to dump profile information.
+  struct sigaction act;
+  act.sa_handler = dump_profile;
+  sigemptyset(&act.sa_mask);
+  act.sa_flags = 0;
+  sigaction(SIGINT, &act, NULL);
+  sigaction(SIGTERM, &act, NULL);
+
+  int mem_size = 256;
+  char* mem = (char*)valloc(mem_size);
+  bzero(mem, mem_size);
+
+  srand48(getpid());
+
+  INetwork* network = nullptr;
+  if (transport_layer == "UDP")
+  {
+    network = Create_UDP_Network().release();
+    LOG_INFO << "Transport: UDP" << std::endl;
+  }
+  else if (transport_layer == "UDP_MT")
+  {
+    network =
+      Create_UDP_Network_MultiThreaded(id % num_receivers_replicas).release();
+    LOG_INFO << "Transport: UDP_MT" << std::endl;
+  }
+  else
+  {
+    LOG_FATAL << "--transport {UDP || UDP_MT}" << std::endl;
+  }
+
+  auto store = std::make_shared<ccf::Store>(
+    pbft::replicate_type_pbft, pbft::replicated_tables_pbft);
+  auto& pbft_requests_map = store->create<pbft::RequestsMap>(
+    pbft::Tables::PBFT_REQUESTS, kv::SecurityDomain::PUBLIC);
+  auto& pbft_pre_prepares_map = store->create<pbft::PrePreparesMap>(
+    pbft::Tables::PBFT_PRE_PREPARES, kv::SecurityDomain::PUBLIC);
+  auto& signatures = store->create<ccf::Signatures>(ccf::Tables::SIGNATURES);
+  auto replica_store =
+    std::make_unique<pbft::Adaptor<ccf::Store, kv::DeserialiseSuccess>>(store);
+
+  int used_bytes = Byz_init_replica(
+    node_info,
+    mem,
+    mem_size,
+    exec_command,
+    network,
+    pbft_requests_map,
+    pbft_pre_prepares_map,
+    signatures,
+    *replica_store,
+    &message_receive_base);
+
+  Byz_start_replica();
+  service_mem = mem + used_bytes;
+  Byz_configure_principals();
+
+  if (with_timeouts)
+  {
+    start_delay_timer();
+  }
+
+  if (test_client_proxy)
+  {
+    setup_client_proxy();
+  }
+
+  Byz_replica_run();
+}
--- a/src/consensus/pbft/libbyz/test/test_ledger_replay.cpp
+++ b/src/consensus/pbft/libbyz/test/test_ledger_replay.cpp
@ -40,7 +40,8 @@ public:
    [this](
      std::array<std::unique_ptr<ExecCommandMsg>, Max_requests_in_batch>& msgs,
      ByzInfo& info,
-      uint32_t num_requests) {
+      uint32_t num_requests,
+      uint64_t nonce) {
      for (uint32_t i = 0; i < num_requests; ++i)
      {
        std::unique_ptr<ExecCommandMsg>& msg = msgs[i];
@ -58,7 +59,7 @@ public:

        outb.contents =
          pbft::GlobalState::get_replica().create_response_message(
-            client, rid, 0);
+            client, rid, 0, nonce);
        outb.size = 0;
        auto request = reinterpret_cast<fake_req*>(inb->contents);
        info.ctx = request->ctx;
@ -285,7 +286,7 @@ TEST_CASE("Test Ledger Replay")
      // request is compatible but pre-prepare root is different
      rqueue.append(request);
      size_t num_requests = 1;
-      auto pp = std::make_unique<Pre_prepare>(1, i, rqueue, num_requests);
+      auto pp = std::make_unique<Pre_prepare>(1, i, rqueue, num_requests, 0);

      // imitate exec command
      ByzInfo info;
--- a/src/consensus/pbft/libbyz/types.h
+++ b/src/consensus/pbft/libbyz/types.h
@ -113,4 +113,5 @@ struct ExecCommandMsg
 using ExecCommand = std::function<int(
  std::array<std::unique_ptr<ExecCommandMsg>, Max_requests_in_batch>& msgs,
  ByzInfo&,
-  uint32_t)>;
+  uint32_t,
+  uint64_t)>;
--- a/src/consensus/pbft/libbyz/view_info.cpp
+++ b/src/consensus/pbft/libbyz/view_info.cpp
@ -142,7 +142,7 @@ void View_info::send_proofs(Seqno n, View vi, int dest)
    {
      if (ri.ods[i].v >= vi)
      {
-        Prepare prep(ri.ods[i].v, n, ri.ods[i].d, p.get());
+        Prepare prep(ri.ods[i].v, n, ri.ods[i].d, 0, p.get());
        pbft::GlobalState::get_node().send(&prep, dest);
      }
    }
--- a/src/consensus/pbft/pbft_config.h
+++ b/src/consensus/pbft/pbft_config.h
@ -55,12 +55,14 @@ namespace pbft
        std::unique_ptr<ExecCommandMsg> msg_,
        ByzInfo& info_,
        PbftConfigCcf* self_,
-        bool is_first_request_) :
+        bool is_first_request_,
+        uint64_t nonce_) :
        msg(std::move(msg_)),
        info(info_),
        self(self_),
        is_first_request(is_first_request_),
-        did_exec_gov_req(false)
+        did_exec_gov_req(false),
+        nonce(nonce_)
      {}

      std::unique_ptr<ExecCommandMsg> msg;
@ -69,6 +71,7 @@ namespace pbft
      PbftConfigCcf* self;
      bool is_first_request;
      bool did_exec_gov_req;
+      uint64_t nonce;
    };

    static void ExecuteCb(std::unique_ptr<enclave::Tmsg<ExecutionCtx>> c)
@ -177,7 +180,7 @@ namespace pbft
      info.ctx = rep.version;

      outb.contents = self->message_receive_base->create_response_message(
-        client, rid, rep.result.size());
+        client, rid, rep.result.size(), execution_ctx.nonce);

      outb.size = rep.result.size();
      auto outb_ptr = (uint8_t*)outb.contents;
@ -205,14 +208,15 @@ namespace pbft
        std::array<std::unique_ptr<ExecCommandMsg>, Max_requests_in_batch>&
          msgs,
        ByzInfo& info,
-        uint32_t num_requests) {
+        uint32_t num_requests,
+        uint64_t nonce) {
        info.pending_cmd_callbacks = num_requests;
        for (uint32_t i = 0; i < num_requests; ++i)
        {
          std::unique_ptr<ExecCommandMsg>& msg = msgs[i];
          uint16_t reply_thread = msg->reply_thread;
          auto execution_ctx = std::make_unique<enclave::Tmsg<ExecutionCtx>>(
-            &Execute, std::move(msg), info, this, is_first_request);
+            &Execute, std::move(msg), info, this, is_first_request, nonce);
          is_first_request = false;

          if (info.cb != nullptr)
--- a/src/tls/entropy.h
+++ b/src/tls/entropy.h
@ -48,6 +48,21 @@ namespace tls
      return data;
    }

+    uint64_t random64() override
+    {
+      uint64_t rnd;
+      uint64_t len = sizeof(uint64_t);
+
+      if (
+        mbedtls_ctr_drbg_random(
+          &drbg, reinterpret_cast<unsigned char*>(&rnd), len) != 0)
+      {
+        throw std::logic_error("Couldn't create random data");
+      }
+
+      return rnd;
+    }
+
    void random(unsigned char* data, size_t len) override
    {
      if (mbedtls_ctr_drbg_random(&drbg, data, len) != 0)
--- a/src/tls/intel_drng.h
+++ b/src/tls/intel_drng.h
@ -34,6 +34,7 @@ namespace tls
    virtual rng_func_t get_rng() = 0;
    virtual std::vector<uint8_t> random(size_t len) = 0;
    virtual void random(unsigned char* data, size_t len) = 0;
+    virtual uint64_t random64() = 0;
    virtual ~Entropy() {}
  };

@ -275,6 +276,19 @@ namespace tls
      return std::vector<uint8_t>(buf, buf + len);
    }

+    uint64_t random64() override
+    {
+      uint64_t rnd;
+      uint64_t len = sizeof(uint64_t);
+
+      if (rdrand_get_bytes(len, reinterpret_cast<unsigned char*>(&rnd)) < len)
+      {
+        throw std::logic_error("Couldn't create random data");
+      }
+
+      return rnd;
+    }
+
    void random(unsigned char* data, size_t len) override
    {
      if (rdrand_get_bytes(len, data) < len)