KEP-1321: Send state requests when swarm checkpoint conflicts with local

checkpoint. Refactor checkpoint handling.

CMakeLists.txt

@@ -85,6 +85,7 @@ add_subdirectory(pbft)
 add_subdirectory(chaos)
 add_subdirectory(crypto)
 add_subdirectory(monitor)
+add_subdirectory(mocks)
 
 include(cmake/static_analysis.cmake)
 

crypto/crypto.cpp

@@ -78,6 +78,10 @@ crypto::extract_payload(const bzn_envelope& msg)
         {
             return msg.status_response();
         }
+        case bzn_envelope::kCheckpointMsg:
+        {
+            return msg.checkpoint_msg();
+        }
         default :
         {
             throw std::runtime_error(

mocks/CMakeLists.txt

@@ -0,0 +1,10 @@
+add_library(smart_mocks STATIC
+        smart_mock_node.hpp
+        smart_mock_node.cpp
+        smart_mock_io.hpp
+        smart_mock_io.cpp
+        )
+
+add_dependencies(smart_mocks boost proto googletest)
+
+target_include_directories(smart_mocks PRIVATE ${PROTO_INCLUDE_DIR})

mocks/smart_mock_io.cpp

@@ -0,0 +1,176 @@
+// Copyright (C) 2019 Bluzelle
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License, version 3,
+// as published by the Free Software Foundation.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+
+#include <mocks/smart_mock_io.hpp>
+
+using namespace ::testing;
+using namespace bzn;
+
+
+bzn::asio::smart_mock_io::smart_mock_io()
+{
+    EXPECT_CALL(*this, make_unique_strand()).WillRepeatedly(Invoke(
+            []()
+            {
+                auto strand = std::make_unique<bzn::asio::Mockstrand_base>();
+                EXPECT_CALL(*strand, wrap(A<bzn::asio::close_handler>())).WillRepeatedly(ReturnArg<0>());
+                EXPECT_CALL(*strand, wrap(A<bzn::asio::read_handler>())).WillRepeatedly(ReturnArg<0>());
+                EXPECT_CALL(*strand, wrap(A<bzn::asio::task>())).WillRepeatedly(ReturnArg<0>());
+                return strand;
+            }));
+
+    EXPECT_CALL(*this, post(_)).WillRepeatedly(Invoke(
+            [&](auto func)
+            {
+                this->real_io_context->post(func);
+            }));
+
+    EXPECT_CALL(*this, make_unique_steady_timer()).WillRepeatedly(Invoke(
+            [&]()
+            {
+                auto id = timer_count++;
+
+                auto timer = std::make_unique<bzn::asio::Mocksteady_timer_base>();
+                EXPECT_CALL(*timer, async_wait(_)).WillRepeatedly(Invoke(
+                        [&, id](auto wh)
+                        {
+                            timer_callbacks[id] = wh;
+                        }));
+                EXPECT_CALL(*timer, expires_from_now(_)).Times(AnyNumber());
+                return timer;
+            }));
+
+    EXPECT_CALL(*this, make_unique_tcp_socket()).WillRepeatedly(Invoke(
+            [&]()
+            {
+                auto id = socket_count++;
+
+                auto mock_socket = std::make_unique<bzn::asio::Mocktcp_socket_base>();
+
+                static boost::asio::io_context io;
+                static boost::asio::ip::tcp::socket socket(io);
+                this->socket_id_map.insert_or_assign(socket.native_handle(), id);
+                EXPECT_CALL(*mock_socket, get_tcp_socket()).WillRepeatedly(ReturnRef(socket));
+
+                EXPECT_CALL(*mock_socket, remote_endpoint()).Times(AnyNumber());
+
+                EXPECT_CALL(*mock_socket, async_connect(_, _)).Times(AtMost(1)).WillOnce(Invoke(
+                        [&](auto, auto handler)
+                        {
+                            this->real_io_context->post(std::bind(handler,
+                                    this->tcp_connect_works ? boost::system::error_code{} : boost::asio::error::connection_refused));
+                        }));
+
+                return mock_socket;
+            }));
+
+    EXPECT_CALL(*this, make_unique_tcp_acceptor(_)).Times(AtMost(1)).WillOnce(Invoke(
+            [&](auto& /*ep*/)
+            {
+                auto mock_acceptor = std::make_unique<bzn::asio::Mocktcp_acceptor_base>();
+
+                EXPECT_CALL(*mock_acceptor, async_accept(_, _)).WillRepeatedly(Invoke(
+                        [&](auto& socket, auto handler)
+                        {
+                            auto id = this->socket_id_map.at(socket.get_tcp_socket().native_handle());
+                            this->tcp_accept_handlers.insert_or_assign(id, handler);
+                        }));
+
+                return mock_acceptor;
+            }));
+
+    EXPECT_CALL(*(this->websocket), make_unique_websocket_stream(_)).WillRepeatedly(Invoke(
+            [&](auto& socket)
+            {
+                auto id = this->socket_id_map.at(socket.native_handle());
+                this->socket_is_open.insert_or_assign(id, true);
+                auto wss = std::make_unique<bzn::beast::Mockwebsocket_stream_base>();
+
+                EXPECT_CALL(*wss, async_accept(_)).Times(AtMost(1)).WillOnce(Invoke(
+                        [&, id](auto handler)
+                        {
+                            this->ws_accept_handlers.insert_or_assign(id, handler);
+                        }));
+
+                EXPECT_CALL(*wss, async_write(_, _)).WillRepeatedly(Invoke(
+                        [&, id](const boost::asio::mutable_buffers_1& buffer, auto handler)
+                        {
+                            this->ws_write_closures.insert_or_assign(id,
+                                    [handler, &buffer]()
+                                    {
+                                        char* raw_buf = static_cast<char*>(buffer.begin()->data());
+                                        std::string result(raw_buf, buffer.begin()->size());
+                                        handler(boost::system::error_code{}, result.size());
+
+                                        return result;
+                                    });
+                        }));
+
+                EXPECT_CALL(*wss, async_read(_, _)).WillRepeatedly(Invoke(
+                        [&, id](auto& buffer, auto handler)
+                        {
+                            this->ws_read_closures.insert_or_assign(id,
+                                    [handler, &buffer](std::string data)
+                                    {
+                                        size_t n = boost::asio::buffer_copy(buffer.prepare(data.size()), boost::asio::buffer(data));
+                                        buffer.commit(n);
+
+                                        handler(boost::system::error_code{}, data.size());
+                                    });
+                        }));
+
+                EXPECT_CALL(*wss, async_handshake(_, _, _)).Times(AtMost(1)).WillRepeatedly(Invoke(
+                        [&](auto, auto, auto handler)
+                        {
+                            this->real_io_context->post(std::bind(handler, boost::system::error_code{}));
+                        }));
+
+                EXPECT_CALL(*wss, is_open()).WillRepeatedly(Invoke(
+                        [&, id]()
+                        {
+                            return this->socket_is_open.at(id);
+                        }));
+
+                this->ws_closed.insert_or_assign(id, false);
+                EXPECT_CALL(*wss, async_close(_, _)).WillRepeatedly(Invoke(
+                        [&, id](auto /*reason*/, auto handler)
+                        {
+                            this->ws_closed.insert_or_assign(id, true);
+                            this->real_io_context->post(std::bind(handler, boost::system::error_code{}));
+                        }));
+
+                EXPECT_CALL(*wss, binary(_)).Times(AnyNumber());
+
+                return wss;
+            }));
+}
+
+void
+bzn::asio::smart_mock_io::do_incoming_connection(size_t id)
+{
+    this->tcp_accept_handlers.at(id)(boost::system::error_code{});
+    this->ws_accept_handlers.at(id)(boost::system::error_code{});
+}
+
+void
+bzn::asio::smart_mock_io::shutdown()
+{
+    // These callbacks are likely to transitively hold a shared pointers to us,
+    // so cleaning them up is necessary for us to be cleaned up
+    this->timer_callbacks.clear();
+    this->ws_read_closures.clear();
+    this->ws_write_closures.clear();
+    this->ws_accept_handlers.clear();
+    this->tcp_accept_handlers.clear();
+
+}

mocks/smart_mock_io.hpp

@@ -0,0 +1,50 @@
+// Copyright (C) 2019 Bluzelle
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License, version 3,
+// as published by the Free Software Foundation.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+
+#pragma once
+
+#include <mocks/mock_boost_asio_beast.hpp>
+
+namespace bzn::asio
+{
+    class smart_mock_io : public Mockio_context_base
+    {
+    public:
+
+        smart_mock_io();
+        void do_incoming_connection(size_t id);
+        void shutdown();
+
+        std::shared_ptr<bzn::asio::io_context_base> real_io_context = std::make_shared<bzn::asio::io_context>();
+        std::shared_ptr<bzn::beast::Mockwebsocket_base> websocket = std::make_shared<bzn::beast::Mockwebsocket_base>();
+
+        size_t timer_count = 0;
+        std::map<size_t, bzn::asio::wait_handler> timer_callbacks;
+
+        size_t socket_count = 0;
+        std::map<size_t, bzn::asio::accept_handler> tcp_accept_handlers;
+        std::map<size_t, bzn::asio::accept_handler> ws_accept_handlers;
+
+        std::map<size_t, std::function<std::string()>> ws_write_closures;
+        std::map<size_t, std::function<void(std::string)>> ws_read_closures;
+
+        std::map<size_t, bool> ws_closed;
+
+        std::map<size_t, bool> socket_is_open;
+
+        std::map<boost::asio::ip::tcp::socket::native_handle_type, size_t> socket_id_map;
+
+        bool tcp_connect_works = true;
+
+    };
+}

mocks/smart_mock_node.cpp

@@ -0,0 +1,44 @@
+// Copyright (C) 2019 Bluzelle
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License, version 3,
+// as published by the Free Software Foundation.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+#include <mocks/smart_mock_node.hpp>
+
+
+using namespace ::testing;
+
+bzn::smart_mock_node::smart_mock_node()
+{
+    EXPECT_CALL(*this, register_for_message(_, _)).WillRepeatedly(Invoke(
+            [&](auto type, auto handler)
+            {
+                if (this->registrants.count(type) > 0)
+                {
+                    throw std::runtime_error("duplicate node registration");
+                }
+
+                this->registrants[type] = handler;
+                return true;
+            }
+            ));
+}
+
+void bzn::smart_mock_node::deliver(const bzn_envelope& msg)
+{
+    if (this->registrants.count(msg.payload_case()) == 0)
+    {
+        throw std::runtime_error("undeliverable message");
+    }
+
+    this->registrants.at(msg.payload_case())(msg, nullptr);
+}

mocks/smart_mock_node.hpp

@@ -0,0 +1,32 @@
+// Copyright (C) 2019 Bluzelle
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License, version 3,
+// as published by the Free Software Foundation.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+#pragma once
+
+#include <mocks/mock_node_base.hpp>
+#include <unordered_map>
+
+namespace bzn
+{
+    class smart_mock_node : public Mocknode_base
+    {
+    public:
+        smart_mock_node();
+
+        void deliver(const bzn_envelope&);
+
+        std::unordered_map<bzn_envelope::PayloadCase, bzn::protobuf_handler> registrants;
+
+    };
+}

node/session.cpp

@@ -74,7 +74,6 @@ session::open(std::shared_ptr<bzn::beast::websocket_base> ws_factory)
 {
     this->strand->wrap([self = shared_from_this(), ws_factory]()
     {
-
         std::shared_ptr<bzn::asio::tcp_socket_base> socket = self->io_context->make_unique_tcp_socket();
         socket->async_connect(self->ep,
             self->strand->wrap([self, socket, ws_factory](const boost::system::error_code& ec)

node/test/CMakeLists.txt

@@ -1,4 +1,4 @@
-set(test_srcs node_test.cpp session_test.cpp node_test_common.hpp)
-set(test_libs node proto options crypto ${Protobuf_LIBRARIES})
+set(test_srcs node_test.cpp session_test.cpp)
+set(test_libs node proto options crypto smart_mocks ${Protobuf_LIBRARIES})
 
 add_gmock_test(node)