Comment on the usecase of the ThriftHandle::Send() method

Co-authored-by: János Benjamin Antal <antaljanosbenjamin@users.noreply.github.com>

cmake/FindSodium.cmake

@@ -0,0 +1,303 @@
+# Written in 2016 by Henrik Steffen Gaßmann <henrik@gassmann.onl>
+#
+# To the extent possible under law, the author(s) have dedicated all
+# copyright and related and neighboring rights to this software to the
+# public domain worldwide. This software is distributed without any warranty.
+#
+# You should have received a copy of the CC0 Public Domain Dedication
+# along with this software. If not, see
+#
+# http://creativecommons.org/publicdomain/zero/1.0/
+#
+# #######################################################################
+# Tries to find the local libsodium installation.
+#
+# On Windows the sodium_DIR environment variable is used as a default
+# hint which can be overridden by setting the corresponding cmake variable.
+#
+# Once done the following variables will be defined:
+#
+# sodium_FOUND
+# sodium_INCLUDE_DIR
+# sodium_LIBRARY_DEBUG
+# sodium_LIBRARY_RELEASE
+#
+#
+# Furthermore an imported "sodium" target is created.
+#
+
+if(CMAKE_C_COMPILER_ID STREQUAL "GNU"
+  OR CMAKE_C_COMPILER_ID STREQUAL "Clang")
+  set(_GCC_COMPATIBLE 1)
+endif()
+
+# static library option
+if(NOT DEFINED sodium_USE_STATIC_LIBS)
+  option(sodium_USE_STATIC_LIBS "enable to statically link against sodium" OFF)
+endif()
+
+if(NOT(sodium_USE_STATIC_LIBS EQUAL sodium_USE_STATIC_LIBS_LAST))
+  unset(sodium_LIBRARY CACHE)
+  unset(sodium_LIBRARY_DEBUG CACHE)
+  unset(sodium_LIBRARY_RELEASE CACHE)
+  unset(sodium_DLL_DEBUG CACHE)
+  unset(sodium_DLL_RELEASE CACHE)
+  set(sodium_USE_STATIC_LIBS_LAST ${sodium_USE_STATIC_LIBS} CACHE INTERNAL "internal change tracking variable")
+endif()
+
+# #######################################################################
+# UNIX
+if(UNIX)
+  # import pkg-config
+  find_package(PkgConfig QUIET)
+
+  if(PKG_CONFIG_FOUND)
+    pkg_check_modules(sodium_PKG QUIET libsodium)
+  endif()
+
+  if(sodium_USE_STATIC_LIBS)
+    foreach(_libname ${sodium_PKG_STATIC_LIBRARIES})
+      if(NOT _libname MATCHES "^lib.*\\.a$") # ignore strings already ending with .a
+        list(INSERT sodium_PKG_STATIC_LIBRARIES 0 "lib${_libname}.a")
+      endif()
+    endforeach()
+
+    list(REMOVE_DUPLICATES sodium_PKG_STATIC_LIBRARIES)
+
+    # if pkgconfig for libsodium doesn't provide
+    # static lib info, then override PKG_STATIC here..
+    if(NOT sodium_PKG_STATIC_FOUND)
+      set(sodium_PKG_STATIC_LIBRARIES libsodium.a)
+    endif()
+
+    set(XPREFIX sodium_PKG_STATIC)
+  else()
+    if(NOT sodium_PKG_FOUND)
+      set(sodium_PKG_LIBRARIES sodium)
+    endif()
+
+    set(XPREFIX sodium_PKG)
+  endif()
+
+  find_path(sodium_INCLUDE_DIR sodium.h
+    HINTS ${${XPREFIX}_INCLUDE_DIRS}
+  )
+  find_library(sodium_LIBRARY_DEBUG NAMES ${${XPREFIX}_LIBRARIES}
+    HINTS ${${XPREFIX}_LIBRARY_DIRS}
+  )
+  find_library(sodium_LIBRARY_RELEASE NAMES ${${XPREFIX}_LIBRARIES}
+    HINTS ${${XPREFIX}_LIBRARY_DIRS}
+  )
+
+# #######################################################################
+# Windows
+elseif(WIN32)
+  set(sodium_DIR "$ENV{sodium_DIR}" CACHE FILEPATH "sodium install directory")
+  mark_as_advanced(sodium_DIR)
+
+  find_path(sodium_INCLUDE_DIR sodium.h
+    HINTS ${sodium_DIR}
+    PATH_SUFFIXES include
+  )
+
+  if(MSVC)
+    # detect target architecture
+    file(WRITE "${CMAKE_CURRENT_BINARY_DIR}/arch.cpp" [=[
+            #if defined _M_IX86
+            #error ARCH_VALUE x86_32
+            #elif defined _M_X64
+            #error ARCH_VALUE x86_64
+            #endif
+            #error ARCH_VALUE unknown
+        ]=])
+    try_compile(_UNUSED_VAR "${CMAKE_CURRENT_BINARY_DIR}" "${CMAKE_CURRENT_BINARY_DIR}/arch.cpp"
+      OUTPUT_VARIABLE _COMPILATION_LOG
+    )
+    string(REGEX REPLACE ".*ARCH_VALUE ([a-zA-Z0-9_]+).*" "\\1" _TARGET_ARCH "${_COMPILATION_LOG}")
+
+    # construct library path
+    if(_TARGET_ARCH STREQUAL "x86_32")
+      string(APPEND _PLATFORM_PATH "Win32")
+    elseif(_TARGET_ARCH STREQUAL "x86_64")
+      string(APPEND _PLATFORM_PATH "x64")
+    else()
+      message(FATAL_ERROR "the ${_TARGET_ARCH} architecture is not supported by Findsodium.cmake.")
+    endif()
+
+    string(APPEND _PLATFORM_PATH "/$$CONFIG$$")
+
+    if(MSVC_VERSION LESS 1900)
+      math(EXPR _VS_VERSION "${MSVC_VERSION} / 10 - 60")
+    else()
+      math(EXPR _VS_VERSION "${MSVC_VERSION} / 10 - 50")
+    endif()
+
+    string(APPEND _PLATFORM_PATH "/v${_VS_VERSION}")
+
+    if(sodium_USE_STATIC_LIBS)
+      string(APPEND _PLATFORM_PATH "/static")
+    else()
+      string(APPEND _PLATFORM_PATH "/dynamic")
+    endif()
+
+    string(REPLACE "$$CONFIG$$" "Debug" _DEBUG_PATH_SUFFIX "${_PLATFORM_PATH}")
+    string(REPLACE "$$CONFIG$$" "Release" _RELEASE_PATH_SUFFIX "${_PLATFORM_PATH}")
+
+    find_library(sodium_LIBRARY_DEBUG libsodium.lib
+      HINTS ${sodium_DIR}
+      PATH_SUFFIXES ${_DEBUG_PATH_SUFFIX}
+    )
+    find_library(sodium_LIBRARY_RELEASE libsodium.lib
+      HINTS ${sodium_DIR}
+      PATH_SUFFIXES ${_RELEASE_PATH_SUFFIX}
+    )
+
+    if(NOT sodium_USE_STATIC_LIBS)
+      set(CMAKE_FIND_LIBRARY_SUFFIXES_BCK ${CMAKE_FIND_LIBRARY_SUFFIXES})
+      set(CMAKE_FIND_LIBRARY_SUFFIXES ".dll")
+      find_library(sodium_DLL_DEBUG libsodium
+        HINTS ${sodium_DIR}
+        PATH_SUFFIXES ${_DEBUG_PATH_SUFFIX}
+      )
+      find_library(sodium_DLL_RELEASE libsodium
+        HINTS ${sodium_DIR}
+        PATH_SUFFIXES ${_RELEASE_PATH_SUFFIX}
+      )
+      set(CMAKE_FIND_LIBRARY_SUFFIXES ${CMAKE_FIND_LIBRARY_SUFFIXES_BCK})
+    endif()
+
+  elseif(_GCC_COMPATIBLE)
+    if(sodium_USE_STATIC_LIBS)
+      find_library(sodium_LIBRARY_DEBUG libsodium.a
+        HINTS ${sodium_DIR}
+        PATH_SUFFIXES lib
+      )
+      find_library(sodium_LIBRARY_RELEASE libsodium.a
+        HINTS ${sodium_DIR}
+        PATH_SUFFIXES lib
+      )
+    else()
+      find_library(sodium_LIBRARY_DEBUG libsodium.dll.a
+        HINTS ${sodium_DIR}
+        PATH_SUFFIXES lib
+      )
+      find_library(sodium_LIBRARY_RELEASE libsodium.dll.a
+        HINTS ${sodium_DIR}
+        PATH_SUFFIXES lib
+      )
+
+      file(GLOB _DLL
+        LIST_DIRECTORIES false
+        RELATIVE "${sodium_DIR}/bin"
+        "${sodium_DIR}/bin/libsodium*.dll"
+      )
+      find_library(sodium_DLL_DEBUG ${_DLL} libsodium
+        HINTS ${sodium_DIR}
+        PATH_SUFFIXES bin
+      )
+      find_library(sodium_DLL_RELEASE ${_DLL} libsodium
+        HINTS ${sodium_DIR}
+        PATH_SUFFIXES bin
+      )
+    endif()
+  else()
+    message(FATAL_ERROR "this platform is not supported by FindSodium.cmake")
+  endif()
+
+# #######################################################################
+# unsupported
+else()
+  message(FATAL_ERROR "this platform is not supported by FindSodium.cmake")
+endif()
+
+# #######################################################################
+# common stuff
+
+# extract sodium version
+if(sodium_INCLUDE_DIR)
+  set(_VERSION_HEADER "${_INCLUDE_DIR}/sodium/version.h")
+
+  if(EXISTS _VERSION_HEADER)
+    file(READ "${_VERSION_HEADER}" _VERSION_HEADER_CONTENT)
+    string(REGEX REPLACE ".*#[ \t]*define[ \t]*SODIUM_VERSION_STRING[ \t]*\"([^\n]*)\".*" "\\1"
+      sodium_VERSION "${_VERSION_HEADER_CONTENT}")
+    set(sodium_VERSION "${sodium_VERSION}" PARENT_SCOPE)
+  endif()
+endif()
+
+# communicate results
+include(FindPackageHandleStandardArgs)
+find_package_handle_standard_args(
+  Sodium # The name must be either uppercase or match the filename case.
+  REQUIRED_VARS
+  sodium_LIBRARY_RELEASE
+  sodium_LIBRARY_DEBUG
+  sodium_INCLUDE_DIR
+  VERSION_VAR
+  sodium_VERSION
+)
+
+if(Sodium_FOUND)
+  set(sodium_LIBRARIES
+    optimized ${sodium_LIBRARY_RELEASE} debug ${sodium_LIBRARY_DEBUG})
+endif()
+
+# mark file paths as advanced
+mark_as_advanced(sodium_INCLUDE_DIR)
+mark_as_advanced(sodium_LIBRARY_DEBUG)
+mark_as_advanced(sodium_LIBRARY_RELEASE)
+
+if(WIN32)
+  mark_as_advanced(sodium_DLL_DEBUG)
+  mark_as_advanced(sodium_DLL_RELEASE)
+endif()
+
+# create imported target
+if(sodium_USE_STATIC_LIBS)
+  set(_LIB_TYPE STATIC)
+else()
+  set(_LIB_TYPE SHARED)
+endif()
+
+if(NOT TARGET sodium)
+  add_library(sodium ${_LIB_TYPE} IMPORTED)
+endif()
+
+set_target_properties(sodium PROPERTIES
+  INTERFACE_INCLUDE_DIRECTORIES "${sodium_INCLUDE_DIR}"
+  IMPORTED_LINK_INTERFACE_LANGUAGES "C"
+)
+
+if(sodium_USE_STATIC_LIBS)
+  set_target_properties(sodium PROPERTIES
+    INTERFACE_COMPILE_DEFINITIONS "SODIUM_STATIC"
+    IMPORTED_LOCATION "${sodium_LIBRARY_RELEASE}"
+    IMPORTED_LOCATION_DEBUG "${sodium_LIBRARY_DEBUG}"
+  )
+else()
+  if(UNIX)
+    set_target_properties(sodium PROPERTIES
+      IMPORTED_LOCATION "${sodium_LIBRARY_RELEASE}"
+      IMPORTED_LOCATION_DEBUG "${sodium_LIBRARY_DEBUG}"
+    )
+  elseif(WIN32)
+    set_target_properties(sodium PROPERTIES
+      IMPORTED_IMPLIB "${sodium_LIBRARY_RELEASE}"
+      IMPORTED_IMPLIB_DEBUG "${sodium_LIBRARY_DEBUG}"
+    )
+
+    if(NOT(sodium_DLL_DEBUG MATCHES ".*-NOTFOUND"))
+      set_target_properties(sodium PROPERTIES
+        IMPORTED_LOCATION_DEBUG "${sodium_DLL_DEBUG}"
+      )
+    endif()
+
+    if(NOT(sodium_DLL_RELEASE MATCHES ".*-NOTFOUND"))
+      set_target_properties(sodium PROPERTIES
+        IMPORTED_LOCATION_RELWITHDEBINFO "${sodium_DLL_RELEASE}"
+        IMPORTED_LOCATION_MINSIZEREL "${sodium_DLL_RELEASE}"
+        IMPORTED_LOCATION_RELEASE "${sodium_DLL_RELEASE}"
+      )
+    endif()
+  endif()
+endif()

cmake/MgThrift.cmake

@@ -0,0 +1,176 @@
+find_package(Gflags)
+find_package(Folly)
+find_package(wangle)
+find_package(fizz)
+find_package(proxygen)
+find_package(FBThrift)
+set(THRIFTCPP2 "FBThrift::thriftcpp2")
+set(THRIFT1 ${FBTHRIFT_COMPILER})
+
+include(${FBTHRIFT_INCLUDE_DIR}/thrift/ThriftLibrary.cmake)
+
+set(MG_INTERFACE_TARGET_NAME_PREFIX "mg-interface")
+
+function(_mg_thrift_generate
+  file_name
+  services
+  language
+  options
+  file_path
+  output_path
+  include_prefix
+)
+  cmake_parse_arguments(THRIFT_GENERATE # Prefix
+    "" # Options
+    "" # One Value args
+    "THRIFT_INCLUDE_DIRECTORIES" # Multi-value args
+    "${ARGN}")
+
+  set(thrift_include_directories)
+
+  foreach(dir ${THRIFT_GENERATE_THRIFT_INCLUDE_DIRECTORIES})
+    list(APPEND thrift_include_directories "-I" "${dir}")
+  endforeach()
+
+  set("${file_name}-${language}-HEADERS"
+    ${output_path}/gen-${language}/${file_name}_constants.h
+    ${output_path}/gen-${language}/${file_name}_data.h
+    ${output_path}/gen-${language}/${file_name}_metadata.h
+    ${output_path}/gen-${language}/${file_name}_types.h
+    ${output_path}/gen-${language}/${file_name}_types.tcc
+  )
+  set("${file_name}-${language}-SOURCES"
+    ${output_path}/gen-${language}/${file_name}_constants.cpp
+    ${output_path}/gen-${language}/${file_name}_data.cpp
+    ${output_path}/gen-${language}/${file_name}_types.cpp
+  )
+
+  if(NOT "${options}" MATCHES "no_metadata")
+    set("${file_name}-${language}-SOURCES"
+      ${${file_name}-${language}-SOURCES}
+      ${output_path}/gen-${language}/${file_name}_metadata.cpp
+    )
+  endif()
+
+  foreach(service ${services})
+    set("${file_name}-${language}-HEADERS"
+      ${${file_name}-${language}-HEADERS}
+      ${output_path}/gen-${language}/${service}.h
+      ${output_path}/gen-${language}/${service}.tcc
+      ${output_path}/gen-${language}/${service}AsyncClient.h
+      ${output_path}/gen-${language}/${service}_custom_protocol.h
+    )
+    set("${file_name}-${language}-SOURCES"
+      ${${file_name}-${language}-SOURCES}
+      ${output_path}/gen-${language}/${service}.cpp
+      ${output_path}/gen-${language}/${service}AsyncClient.cpp
+    )
+  endforeach()
+
+  if("${include_prefix}" STREQUAL "")
+    set(include_prefix_text "")
+  else()
+    set(include_prefix_text "include_prefix=${include_prefix}")
+
+    if(NOT "${options}" STREQUAL "")
+      set(include_prefix_text ",${include_prefix_text}")
+    endif()
+  endif()
+
+  set(gen_language ${language})
+
+  if("${language}" STREQUAL "cpp2")
+    set(gen_language "mstch_cpp2")
+  elseif("${language}" STREQUAL "py3")
+    set(gen_language "mstch_py3")
+    file(WRITE "${output_path}/gen-${language}/${file_name}/__init__.py")
+  endif()
+
+  add_custom_command(
+    OUTPUT ${${file_name}-${language}-HEADERS}
+    ${${file_name}-${language}-SOURCES}
+    COMMAND ${THRIFT1}
+    --gen "${gen_language}:${options}${include_prefix_text}"
+    -o ${output_path}
+    ${thrift_include_directories}
+    "${file_path}/${file_name}.thrift"
+    DEPENDS
+    ${THRIFT1}
+    "${file_path}/${file_name}.thrift"
+    COMMENT "Generating ${MG_INTERFACE_TARGET_NAME_PREFIX}-${file_name}-${language} files. Output: ${output_path}"
+  )
+  add_custom_target(
+    ${MG_INTERFACE_TARGET_NAME_PREFIX}-${file_name}-${language}-target ALL
+    DEPENDS ${${language}-${language}-HEADERS}
+    ${${file_name}-${language}-SOURCES}
+  )
+
+  set("${file_name}-${language}-SOURCES" ${${file_name}-${language}-SOURCES} PARENT_SCOPE)
+  install(
+    DIRECTORY gen-${language}
+    DESTINATION include/${include_prefix}
+    FILES_MATCHING PATTERN "*.h")
+  install(
+    DIRECTORY gen-${language}
+    DESTINATION include/${include_prefix}
+    FILES_MATCHING PATTERN "*.tcc")
+endfunction()
+
+function(_mg_thrift_object
+  file_name
+  services
+  language
+  options
+  file_path
+  output_path
+  include_prefix
+)
+  _mg_thrift_generate(
+    "${file_name}"
+    "${services}"
+    "${language}"
+    "${options}"
+    "${file_path}"
+    "${output_path}"
+    "${include_prefix}"
+    "${ARGN}"
+  )
+  bypass_source_check(${${file_name}-${language}-SOURCES})
+  add_library(
+    "${MG_INTERFACE_TARGET_NAME_PREFIX}-${file_name}-${language}-obj"
+    OBJECT
+    ${${file_name}-${language}-SOURCES}
+  )
+  add_dependencies(
+    "${MG_INTERFACE_TARGET_NAME_PREFIX}-${file_name}-${language}-obj"
+    "${MG_INTERFACE_TARGET_NAME_PREFIX}-${file_name}-${language}-target"
+  )
+  target_include_directories(${MG_INTERFACE_TARGET_NAME_PREFIX}-${file_name}-${language}-obj PUBLIC ${output_path})
+  message(STATUS "MgThrift will create the Object file : ${MG_INTERFACE_TARGET_NAME_PREFIX}-${file_name}-${language}-obj")
+endfunction()
+
+function(mg_thrift_library
+  file_name
+  services
+  file_path
+  output_path
+  include_prefix
+)
+  _mg_thrift_object(
+    "${file_name}"
+    "${services}"
+    "cpp2"
+    "stack_arguments" # options
+    "${file_path}"
+    "${output_path}"
+    "${include_prefix}"
+    THRIFT_INCLUDE_DIRECTORIES "${FBTHRIFT_INCLUDE_DIR}"
+  )
+  set(LIBRARY_NAME "${MG_INTERFACE_TARGET_NAME_PREFIX}-${file_name}-cpp2")
+  add_library(
+    "${LIBRARY_NAME}"
+    $<TARGET_OBJECTS:${LIBRARY_NAME}-obj>
+  )
+  target_link_libraries("${LIBRARY_NAME}" ${THRIFTCPP2})
+  message("MgThrift will create the library file : ${LIBRARY_NAME}")
+endfunction()

environment/toolchain/fbthrift.patch

@@ -0,0 +1,12 @@
+diff -ur a/thrift/lib/cpp2/server/IOWorkerContext.h b/thrift/lib/cpp2/server/IOWorkerContext.h
+--- a/thrift/lib/cpp2/server/IOWorkerContext.h  2022-06-08 11:50:43.043948657 +0200
++++ b/thrift/lib/cpp2/server/IOWorkerContext.h  2022-06-08 11:47:33.232695125 +0200
+@@ -59,7 +59,7 @@
+           auto aliveLocked = alive->rlock();
+           if (*aliveLocked) {
+             // IOWorkerContext is still alive and so is replyQueue_
+-            queue->startConsumingInternal(&evb);
++            queue->startConsuming(&evb);
+           }
+         });
+   }

src/CMakeLists.txt

@@ -1,6 +1,8 @@
 # CMake configuration for the main memgraph library and executable
+include(MgThrift)
 
 # add memgraph sub libraries, ordered by dependency
+add_subdirectory(interface)
 add_subdirectory(lisp)
 add_subdirectory(utils)
 add_subdirectory(requests)
@@ -18,6 +20,7 @@ add_subdirectory(query/v2)
 add_subdirectory(slk)
 add_subdirectory(rpc)
 add_subdirectory(auth)
+add_subdirectory(coordinator)
 
 if (MG_ENTERPRISE)
   add_subdirectory(audit)
@@ -40,7 +43,7 @@ set(mg_single_node_v2_sources
 )
 
 set(mg_single_node_v2_libs stdc++fs Threads::Threads
-    telemetry_lib mg-query mg-communication mg-memory mg-utils mg-auth mg-license mg-settings)
+    telemetry_lib mg-query mg-communication mg-memory mg-utils mg-auth mg-license mg-settings FBThrift::thriftcpp2)
 if (MG_ENTERPRISE)
   # These are enterprise subsystems
   set(mg_single_node_v2_libs ${mg_single_node_v2_libs} mg-audit)

src/coordinator/CMakeLists.txt

@@ -0,0 +1,10 @@
+set(coordinator_src_files
+    coordinator.hpp
+    shard_map.hpp
+    hybrid_logical_clock.hpp)
+
+find_package(fmt REQUIRED)
+find_package(Threads REQUIRED)
+
+add_library(mg-coordinator STATIC ${coordinator_src_files})
+target_link_libraries(mg-coordinator stdc++fs Threads::Threads fmt::fmt mg-utils)

src/coordinator/coordinator.hpp

@@ -0,0 +1,243 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#pragma once
+
+#include <unordered_set>
+
+#include "coordinator/hybrid_logical_clock.hpp"
+#include "coordinator/shard_map.hpp"
+#include "io/simulator/simulator.hpp"
+#include "io/time.hpp"
+#include "io/transport.hpp"
+
+namespace memgraph::coordinator {
+
+using Address = memgraph::io::Address;
+using SimT = memgraph::io::simulator::SimulatorTransport;
+
+struct HlcRequest {
+  Hlc last_shard_map_version;
+};
+
+struct HlcResponse {
+  Hlc new_hlc;
+  std::optional<ShardMap> fresher_shard_map;
+};
+
+struct GetShardMapRequest {
+  // No state
+};
+
+struct GetShardMapResponse {
+  ShardMap shard_map;
+};
+
+struct AllocateHlcBatchRequest {
+  Hlc low;
+  Hlc high;
+};
+
+struct AllocateHlcBatchResponse {
+  bool success;
+  Hlc low;
+  Hlc high;
+};
+
+struct AllocateEdgeIdBatchRequest {
+  size_t batch_size;
+};
+
+struct AllocateEdgeIdBatchResponse {
+  uint64_t low;
+  uint64_t high;
+};
+
+struct SplitShardRequest {
+  Hlc previous_shard_map_version;
+  Label label;
+  CompoundKey split_key;
+};
+
+struct SplitShardResponse {
+  bool success;
+};
+
+struct RegisterStorageEngineRequest {
+  Address address;
+};
+
+struct RegisterStorageEngineResponse {
+  bool success;
+};
+
+struct DeregisterStorageEngineRequest {
+  Address address;
+};
+
+struct DeregisterStorageEngineResponse {
+  bool success;
+};
+
+struct InitializeLabelRequest {
+  std::string label_name;
+  Hlc last_shard_map_version;
+};
+
+struct InitializeLabelResponse {
+  bool success;
+  std::optional<ShardMap> fresher_shard_map;
+};
+
+using WriteRequests =
+    std::variant<AllocateHlcBatchRequest, AllocateEdgeIdBatchRequest, SplitShardRequest, RegisterStorageEngineRequest,
+                 DeregisterStorageEngineRequest, InitializeLabelRequest>;
+using WriteResponses =
+    std::variant<AllocateHlcBatchResponse, AllocateEdgeIdBatchResponse, SplitShardResponse,
+                 RegisterStorageEngineResponse, DeregisterStorageEngineResponse, InitializeLabelResponse>;
+
+using ReadRequests = std::variant<HlcRequest, GetShardMapRequest>;
+using ReadResponses = std::variant<HlcResponse, GetShardMapResponse>;
+
+class Coordinator {
+  ShardMap shard_map_;
+  /// The highest reserved timestamp / highest allocated timestamp
+  /// is a way for minimizing communication involved in query engines
+  /// reserving Hlc's for their transaction processing.
+  /// Periodically, the coordinator will allocate a batch of timestamps
+  /// and this will need to go over consensus. From that point forward,
+  /// each timestamp in that batch can be given out to "readers" who issue
+  /// HlcRequest without blocking on consensus first. But if
+  /// highest_allocated_timestamp_ approaches highest_reserved_timestamp_,
+  /// it is time to allocate another batch, so that we can keep guaranteeing
+  /// forward progress.
+  /// Any time a coordinator becomes a new leader, it will need to issue
+  /// a new AllocateHlcBatchRequest to create a pool of IDs to allocate.
+  uint64_t highest_allocated_timestamp_;
+  uint64_t highest_reserved_timestamp_;
+
+  /// Query engines need to periodically request batches of unique edge IDs.
+  uint64_t highest_allocated_edge_id_;
+
+  /// Increment our
+  ReadResponses HandleRead(HlcRequest &&hlc_request) {
+    HlcResponse res{};
+
+    auto hlc_shard_map = shard_map_.GetHlc();
+
+    MG_ASSERT(!(hlc_request.last_shard_map_version.logical_id > hlc_shard_map.logical_id));
+
+    res.new_hlc = shard_map_.IncrementShardMapVersion();
+
+    // res.fresher_shard_map = hlc_request.last_shard_map_version.logical_id < hlc_shard_map.logical_id
+    //                             ? std::make_optional(shard_map_)
+    //                             : std::nullopt;
+
+    // Allways return fresher shard_map for now.
+    res.fresher_shard_map = std::make_optional(shard_map_);
+
+    return res;
+  }
+
+  ReadResponses HandleRead(GetShardMapRequest &&get_shard_map_request) {
+    GetShardMapResponse res;
+    res.shard_map = shard_map_;
+    return res;
+  }
+
+  WriteResponses ApplyWrite(AllocateHlcBatchRequest &&ahr) {
+    AllocateHlcBatchResponse res{};
+
+    return res;
+  }
+
+  WriteResponses ApplyWrite(AllocateEdgeIdBatchRequest &&ahr) {
+    AllocateEdgeIdBatchResponse res{};
+
+    uint64_t low = highest_allocated_edge_id_;
+
+    highest_allocated_edge_id_ += ahr.batch_size;
+
+    uint64_t high = highest_allocated_edge_id_;
+
+    res.low = low;
+    res.high = high;
+
+    return res;
+  }
+
+  /// This splits the shard immediately beneath the provided
+  /// split key, keeping the assigned peers identical for now,
+  /// but letting them be gradually migrated over time.
+  WriteResponses ApplyWrite(SplitShardRequest &&split_shard_request) {
+    SplitShardResponse res{};
+
+    if (split_shard_request.previous_shard_map_version != shard_map_.shard_map_version) {
+      res.success = false;
+    } else {
+      res.success = shard_map_.SplitShard(split_shard_request.previous_shard_map_version, split_shard_request.label,
+                                          split_shard_request.split_key);
+    }
+
+    return res;
+  }
+
+  /// This adds the provided storage engine to the standby storage engine pool,
+  /// which can be used to rebalance storage over time.
+  WriteResponses ApplyWrite(RegisterStorageEngineRequest &&register_storage_engine_request) {
+    RegisterStorageEngineResponse res{};
+    // TODO
+
+    return res;
+  }
+
+  /// This begins the process of draining the provided storage engine from all raft
+  /// clusters that it might be participating in.
+  WriteResponses ApplyWrite(DeregisterStorageEngineRequest &&register_storage_engine_request) {
+    DeregisterStorageEngineResponse res{};
+    // TODO
+    // const Address &address = register_storage_engine_request.address;
+    // storage_engine_pool_.erase(address);
+    // res.success = true;
+
+    return res;
+  }
+
+  WriteResponses ApplyWrite(InitializeLabelRequest &&initialize_label_request) {
+    InitializeLabelResponse res{};
+
+    bool success = shard_map_.InitializeNewLabel(initialize_label_request.label_name,
+                                                 initialize_label_request.last_shard_map_version);
+
+    if (success) {
+      res.fresher_shard_map = shard_map_;
+      res.success = false;
+    } else {
+      res.fresher_shard_map = std::nullopt;
+      res.success = true;
+    }
+
+    return res;
+  }
+
+ public:
+  explicit Coordinator(ShardMap sm) : shard_map_{(sm)} {}
+
+  ReadResponses Read(ReadRequests requests) {
+    return std::visit([&](auto &&requests) { return HandleRead(std::move(requests)); }, std::move(requests));
+  }
+
+  WriteResponses Apply(WriteRequests requests) {
+    return std::visit([&](auto &&requests) { return ApplyWrite(std::move(requests)); }, std::move(requests));
+  }
+};
+
+}  // namespace memgraph::coordinator

src/coordinator/hybrid_logical_clock.hpp

@@ -0,0 +1,30 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#pragma once
+
+#include "io/time.hpp"
+
+namespace memgraph::coordinator {
+
+using Time = memgraph::io::Time;
+
+/// Hybrid-logical clock
+struct Hlc {
+  uint64_t logical_id;
+  Time coordinator_wall_clock;
+
+  bool operator==(const Hlc &other) const {
+    return (logical_id == other.logical_id) && (coordinator_wall_clock == other.coordinator_wall_clock);
+  }
+};
+
+}  // namespace memgraph::coordinator

src/coordinator/shard_map.hpp

@@ -0,0 +1,135 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#pragma once
+
+#include <map>
+#include <vector>
+
+#include "coordinator/hybrid_logical_clock.hpp"
+#include "io/address.hpp"
+#include "storage/v3/property_value.hpp"
+
+namespace memgraph::coordinator {
+
+enum class Status : uint8_t {
+  CONSENSUS_PARTICIPANT,
+  INITIALIZING,
+  // TODO(tyler) this will possibly have more states,
+  // depending on the reconfiguration protocol that we
+  // implement.
+};
+
+struct AddressAndStatus {
+  memgraph::io::Address address;
+  Status status;
+};
+
+using memgraph::io::Address;
+
+using CompoundKey = std::vector<memgraph::storage::v3::PropertyValue>;
+using Shard = std::vector<AddressAndStatus>;
+using Shards = std::map<CompoundKey, Shard>;
+
+// use string for intermachine communication and NameIdMapper within the machine
+using Label = std::string;
+
+struct ShardMap {
+  Hlc shard_map_version;
+  std::map<Label, Shards> shards;
+
+  // TODO(gabor) later we will want to update the wallclock time with
+  // the given Io<impl>'s time as well
+  Hlc IncrementShardMapVersion() noexcept {
+    ++shard_map_version.logical_id;
+    return shard_map_version;
+  }
+
+  Hlc GetHlc() const noexcept { return shard_map_version; }
+
+  bool SplitShard(Hlc previous_shard_map_version, Label label, CompoundKey key) {
+    if (CompareShardMapVersions(previous_shard_map_version, shard_map_version)) {
+      MG_ASSERT(shards.contains(label));
+      auto &shards_in_map = shards[label];
+      MG_ASSERT(!shards_in_map.contains(key));
+
+      // Finding the Shard that the new CompoundKey should map to.
+      Shard shard_to_map_to;
+      CompoundKey prev_key = ((*shards_in_map.begin()).first);
+
+      for (auto iter = std::next(shards_in_map.begin()); iter != shards_in_map.end(); ++iter) {
+        const auto &current_key = (*iter).first;
+        if (key > prev_key && key < current_key) {
+          shard_to_map_to = shards_in_map[prev_key];
+        }
+
+        prev_key = (*iter).first;
+      }
+
+      // Apply the split
+      shards_in_map[key] = shard_to_map_to;
+
+      return true;
+    }
+
+    return false;
+  }
+
+  bool InitializeNewLabel(std::string label_name, Hlc last_shard_map_version) {
+    if (shard_map_version != last_shard_map_version) {
+      return false;
+    }
+
+    if (shards.contains(label_name)) {
+      return false;
+    }
+
+    shards.emplace(label_name, Shards{});
+
+    IncrementShardMapVersion();
+
+    return true;
+  }
+
+  void AddServer(Address server_address) {
+    // Find a random place for the server to plug in
+  }
+
+  std::map<Label, Shards> &GetShards() noexcept { return shards; }
+
+  Shards GetShardsForRange(Label label, CompoundKey start, CompoundKey end);
+
+  Shard GetShardForKey(Label label, CompoundKey key) {
+    auto shard_for_label = shards.at(label);
+
+    auto max = (--shard_for_label.end())->first;
+
+    if (key > max) {
+      return shard_for_label[max];
+    }
+
+    for (auto it = shard_for_label.lower_bound(key);; --it) {
+      MG_ASSERT(it->first <= key);
+      return it->second;
+    }
+
+    MG_ASSERT(false, "failed to find shard with a key that is less than or equal to the provided key");
+  }
+
+ private:
+  // TODO(gabor) later we will want to update the wallclock time with
+  // the given Io<impl>'s time as well. This function should just be
+  // replaced with operator== since it is already overloaded for Hlc
+  // objects.
+  bool CompareShardMapVersions(Hlc one, Hlc two) { return one.logical_id == two.logical_id; }
+};
+
+}  // namespace memgraph::coordinator

src/interface/.gitignore

@@ -0,0 +1 @@
+gen-cpp2

src/interface/CMakeLists.txt

@@ -0,0 +1,43 @@
+include(MgThrift)
+
+set(MG_INTERFACE_PATH ${CMAKE_CURRENT_SOURCE_DIR})
+
+mg_thrift_library(
+  "storage" #file_name
+  "Storage" #services
+  "${MG_INTERFACE_PATH}" #file_path
+  "${MG_INTERFACE_PATH}" #output_path
+  "interface" #include_prefix
+)
+
+mg_thrift_library(
+  "meta" #file_name
+  "Meta" #services
+  "${MG_INTERFACE_PATH}" #file_path
+  "${MG_INTERFACE_PATH}" #output_path
+  "interface" #include_prefix
+)
+
+mg_thrift_library(
+  "trial" #file_name
+  "PingPong" #services
+  "${MG_INTERFACE_PATH}" #file_path
+  "${MG_INTERFACE_PATH}" #output_path
+  "interface" #include_prefix
+)
+
+mg_thrift_library(
+  "echo" #file_name
+  "Echo" #services
+  "${MG_INTERFACE_PATH}" #file_path
+  "${MG_INTERFACE_PATH}" #output_path
+  "interface" #include_prefix
+)
+
+mg_thrift_library(
+  "ubermessage" #file_name
+  "UberServer" #services
+  "${MG_INTERFACE_PATH}" #file_path
+  "${MG_INTERFACE_PATH}" #output_path
+  "interface" #include_prefix
+)

src/interface/address.thrift

@@ -0,0 +1,5 @@
+struct Address{
+    1: string unique_id;
+    2: string last_known_ip;
+    3: i32 last_known_port;
+}

src/interface/counter.thrift

@@ -0,0 +1,22 @@
+namespace cpp2 interface.counter
+// https://stackoverflow.com/a/34234874/6639989
+
+struct GetLatestReqeuest {
+  1: optional i64 proposed_value;
+}
+
+struct GetLatestResponse {
+  1: i64 value;
+}
+
+union CounterRequest {
+  1: GetLatestReqeuest get_latest;
+}
+
+union CounterResponse {
+  1: GetLatestResponse get_latest;
+}
+
+service Counter {
+  CounterResponse Request(1: CounterRequest req)
+}

src/interface/echo.thrift

@@ -0,0 +1,20 @@
+struct EchoMessage {
+    1: binary message;
+}
+
+//struct Address{
+//    1: string unique_id;
+//    2: string last_known_ip;
+//    3: i32 last_known_port;
+//}
+
+struct CompoundMessage{
+//    1: Address to_address
+//    2: Address from_address
+    3: binary message
+}
+
+service Echo {
+    oneway void ReceiveSend(1: EchoMessage m)
+    oneway void RecieveCompoundThriftMessage(1: CompoundMessage m)
+}

src/interface/echo_trial.thrift

@@ -0,0 +1,7 @@
+struct EchoMessage {
+    1: binary message;
+}
+
+service Echo {
+    oneway void ReceiveSend(1: EchoMessage m)
+}

src/interface/meta.thrift

@@ -0,0 +1,74 @@
+namespace cpp2 interface.meta
+
+typedef i64 LabelId
+typedef i64 IndexId
+
+struct Result {
+    1: bool success;
+}
+
+struct CreatePrimaryLabelRequest {
+    1: binary name;
+    2: list<binary> primary_keys;
+}
+
+struct CreateLabelResponse {
+    1: Result result;
+    2: LabelId label_id;
+}
+
+struct GetLabelInfosRequest {
+    //  The empty list means return all of the label infos
+    1: list<binary> label_names;
+}
+
+struct LabelInfo {
+    1: binary name;
+    2: LabelId label_id;
+    3: list<binary> primary_keys;
+}
+
+struct GetLabelInfoResponse {
+    1: Result result;
+    2: LabelInfo label_info;
+}
+
+struct GetLabelInfosResponse {
+    1: Result result;
+    2: list<LabelInfo> label_infos;
+}
+
+struct CreateIndexRequest {
+    1: LabelId label_id;
+    2: list<binary> property_names;
+}
+
+struct CreateIndexResponse {
+    1: Result result;
+    2: IndexId index_id;
+}
+
+struct IndexInfo {
+    1: LabelId label_id;
+    2: list<binary> property_names;
+}
+
+struct GetIndexInfosResponse {
+    1: Result result;
+    2: list<IndexInfo> index_infos;
+}
+
+
+service Meta {
+    CreateLabelResponse createPrimaryLabel(1: CreatePrimaryLabelRequest req);
+    CreateLabelResponse createSecondaryLabel(1: binary label_name);
+    Result dropLabel(1: binary label_name);
+    GetLabelInfoResponse getLabelInfo(1:binary label_name);
+    GetLabelInfosResponse getLabelInfos(1: list<binary> label_names);
+
+    CreateIndexResponse createIndex(1: CreateIndexRequest req);
+    // Don't have index names, and the label doesn't identify the index uniquely, therefore
+    Result dropIndex(1: IndexId index_id);
+    GetIndexInfosResponse getIndexInfos();
+    GetIndexInfosResponse getIndexInfosForLabel(1: LabelId label_id);
+}

src/interface/storage.thrift

@@ -0,0 +1,273 @@
+namespace cpp2 interface.storage
+// https://stackoverflow.com/a/34234874/6639989
+
+cpp_include "storage/v2/view.hpp"
+
+typedef i64 VertexId
+typedef i64 Gid
+
+// TODO(antaljanosbenjamin): Use this after introducing 128 bit vertex ids
+// struct VertexId {
+//     1: i64 upper_half;
+//     2: i64 lower_half;
+// }
+
+struct Label {
+    1: i64 id;
+}
+
+struct EdgeType {
+    1: binary name;
+}
+
+struct EdgeId {
+    1: VertexId src;
+    // QUESTION(antaljanosbenjamin): is it okay to have vertex based (edge id = vertex id + edge id inside vertex)?
+    2: Gid gid;
+}
+
+struct Date {
+    1: i16 year;
+    2: byte month;
+    3: byte day;
+}
+
+struct LocalTime {
+    1: byte hour;
+    2: byte minute;
+    3: byte second;
+    4: i16 millisecond;
+    5: i16 microsecond;
+}
+
+struct LocalDateTime {
+    1: Date date;
+    2: LocalTime local_time;
+}
+
+struct Duration {
+    1: i64 milliseconds;
+}
+
+union Value {
+    1: Null null_v;
+    2: bool bool_v;
+    3: i64 int_v;
+    4: double double_v;
+    5: binary string_v;
+    6: list<Value> list_v;
+    7: map<binary, Value> (cpp.template = "std::unordered_map") map_v (cpp2.ref_type = "unique");
+    8: Vertex vertex_v (cpp2.ref_type = "unique");
+    9: Edge edge_v (cpp2.ref_type = "unique");
+    10: Path path_v (cpp2.ref_type = "unique");
+    11: Date date_v;
+    12: LocalTime local_time_v;
+    13: LocalDateTime local_date_time_v;
+    14: Duration duration_v;
+}
+
+struct Null {
+}
+
+struct Vertex {
+    1: VertexId id;
+    // TODO(antaljanosbenjamin): Change to sperate primary and secondary labels when schema is implemented
+    2: list<Label> labels;
+}
+
+struct Edge {
+    1: VertexId src;
+    2: VertexId dst;
+    3: EdgeType type;
+}
+
+struct PathPart {
+    1: Vertex dst;
+    2: Edge edge;
+}
+
+struct Path {
+    1: Vertex src;
+    2: list<PathPart> parts;
+}
+
+struct ValuesMap {
+    1: map<i64, Value> (cpp.template = "std::unordered_map") values_map;
+}
+
+struct MappedValues {
+    1: list<ValuesMap> properties;
+}
+
+struct ListedValues {
+    1: list<list<Value>> properties;
+}
+
+union Values {
+    // This struct is necessary because depending on the request the response
+    // has two different formats:
+    // 1. When the request specifies the returned properties, then they are
+    //    returned in that order, therefore no extra mapping is necessary.
+    // 2. When the request doesn't specify the returned properties, then all
+    //    of the properties are returned. In this case the `mapped` field is
+    //    used. To extract the <key,value> pairs from this struct the
+    //    mapping of i64 -> property name has to be used.
+    1: ListedValues listed;
+    2: MappedValues mapped;
+}
+
+struct Expression {
+    1: binary alias;
+    2: binary expression;
+}
+
+struct Filter {
+    1: binary filter_expression;
+}
+
+enum OrderingDirection {
+    ASCENDING = 1;
+    DESCENDING = 2;
+}
+
+struct OrderBy {
+    1: Expression expression;
+    2: OrderingDirection direction;
+}
+
+struct Result {
+    // Just placeholder data for now
+    1: bool success;
+}
+
+enum View {
+    OLD = 0,
+    NEW = 1
+}  (cpp.enum_strict, cpp.type = "memgraph::storage::View")
+
+struct ScanVerticesRequest {
+    1: i64 transaction_id;
+    2: optional i64 start_id;
+    // Special values are accepted:
+    // * __mg__id (Vertex, but without labels)
+    // * __mg__labels (Vertex, but without the id)
+    // If both of them is specified, then it will result in a single, fully populated vertex
+    // QUESTION(antaljanosbenjamin): Does the `__mg__labels` is necessary? What about passing the `labels` function
+    //                               as an expression? Maybe it is an optimization. For communicating the vertex id
+    //                               the Vertex struct is really handy.
+    3: optional list<binary> props_to_return;
+    4: list<Expression> expressions;
+    5: optional i64 limit;
+    6: View view;
+    7: optional Filter filter;
+}
+
+struct ScanVerticesResponse {
+    1: Result result;
+    2: Values values;
+    3: optional map<i64, binary> (cpp.template = "std::unordered_map") property_name_map;
+    // contains the next start_id if there is any
+    4: optional VertexId next_start_id;
+}
+
+union VertexOrEdgeIds {
+    1: list<VertexId> vertex_ids;
+    2: list<EdgeId> edge_ids;
+}
+
+struct GetPropertiesRequest {
+    1:  i64 transaction_id;
+    2:  VertexOrEdgeIds vertex_or_edge_ids;
+    3:  list<binary> property_names;
+    4:  list<Expression> expressions;
+    5:  bool only_unique = false;
+    6:  optional list<OrderBy> order_by;
+    7:  optional i64 limit;
+    8:  optional Filter filter;
+}
+
+struct GetPropertiesResponse {
+    1: Values values;
+    2: optional map<i64, binary> (cpp.template = "std::unordered_map") property_name_map;
+}
+
+enum EdgeDirection {
+    OUT = 1;
+    IN = 2;
+    BOTH = 3;
+}
+
+struct ExpandOneRequest {
+    1:  i64 transaction_id;
+    2:  list<VertexId> src_vertices;
+    3:  list<EdgeType> edge_types;
+    4:  EdgeDirection direction;
+    5:  bool only_unique_neighbor_rows = false;
+    //  The empty optional means return all of the properties, while an empty
+    //  list means do not return any properties
+    //  TODO(antaljanosbenjamin): All of the special values should be communicated through a single vertex object
+    //                            after schema is implemented
+    //  Special values are accepted:
+    //  * __mg__labels
+    6:  optional list<binary> src_vertex_properties;
+    //  TODO(antaljanosbenjamin): All of the special values should be communicated through a single vertex object
+    //                            after schema is implemented
+    //  Special values are accepted:
+    //  * __mg__dst_id (Vertex, but without labels)
+    //  * __mg__type (binary)
+    7:  optional list<binary> edge_properties;
+    //  QUESTION(antaljanosbenjamin): Maybe also add possibility to expressions evaluated on the source vertex?
+    //  List of expressions evaluated on edges
+    8:  list<Expression> expressions;
+    9:  optional list<OrderBy> order_by;
+    10:  optional i64 limit;
+    11: optional Filter filter;
+}
+
+struct ExpandOneResultRow {
+    // NOTE: This struct could be a single Values with columns something like this:
+    // src_vertex(Vertex), vertex_prop1(Value), vertex_prop2(Value), edges(list<Value>)
+    // where edges might be a list of:
+    // 1. list<Value> if only a defined list of edge properties are returned
+    // 2. map<binary, Value> if all of the edge properties are returned
+    // The drawback of this is currently the key of the map is always interpreted as a string in Value, not as an
+    // integer, which should be in case of mapped properties.
+    1: Vertex src_vertex;
+    2: optional Values src_vertex_properties;
+    3: Values edges;
+}
+
+struct ExpandOneResponse {
+    // This approach might not suit the expand with per shard parrallelization,
+    // because the property_name_map has to be accessed from multiple threads
+    // in order to avoid duplicated keys (two threads might map the same
+    // property with different numbers) and multiple passes (to unify the
+    // mapping amond result returned from different shards).
+    1: list<ExpandOneResultRow> result;
+    2: optional map<i64, binary>  (cpp.template = "std::unordered_map") property_name_map;
+}
+
+struct NewVertex {
+    1: list<i64> label_ids;
+    2: map<i64, Value> properties;
+}
+
+struct CreateVerticesRequest {
+    1: required i64 transaction_id;
+    2: map<i64, binary> (cpp.template = "std::unordered_map") labels_name_map;
+    3: map<i64, binary> (cpp.template = "std::unordered_map") property_name_map;
+    4: list<NewVertex> new_vertices;
+}
+
+
+service Storage {
+    i64 startTransaction()
+    Result commitTransaction(1: i64 transaction_id)
+    void abortTransaction(1: i64 transaction_id)
+
+    Result createVertices(1: CreateVerticesRequest req)
+    ScanVerticesResponse scanVertices(1: ScanVerticesRequest req)
+    GetPropertiesResponse getProperties(1: GetPropertiesRequest req)
+    ExpandOneResponse expandOne(1: ExpandOneRequest req)
+
+}

src/interface/trial.thrift

@@ -0,0 +1,20 @@
+struct Ping {
+    1: binary message;
+}
+
+struct Pong{
+    1: binary message;
+}
+
+struct ValueToAdd{
+    1:  i64 val;
+}
+
+struct ValueToAddResopnse{
+    1:  i64 new_val;
+}
+
+service PingPong {
+    Pong ping(1: Ping req)
+    ValueToAddResopnse AddValue(1: ValueToAdd req)
+}

src/interface/ubermessage.thrift

@@ -0,0 +1,56 @@
+//include "address.thrift"
+
+// TODO(gvolfing) remove this once the include problem is resolved
+struct Address{
+    1: string unique_id;
+    2: string last_known_ip;
+    3: i32 last_known_port;
+}
+
+struct HeartbeatRequest {
+  1: bool test;
+}
+
+struct HeartbeatResponse {
+  1: bool test;
+}
+
+struct ScanAllRequest{
+  1: bool test;
+}
+
+struct ScanAllResponse {
+  1: bool test;
+}
+
+union ToStorageEngine{
+  1: ScanAllRequest scan_all_request;
+  2: HeartbeatRequest heartbeat_request;
+}
+
+union ToQueryEngine {
+  1: ScanAllResponse scan_all_response;
+}
+
+union ToCoordinator {
+  1: HeartbeatResponse heartbeat_response;
+}
+
+union HighLevelUnion {
+    1: ToStorageEngine to_storage_engine;
+    2: ToQueryEngine to_query_engine;
+    3: ToCoordinator to_coordinator;
+}
+
+struct UberMessage {
+    //1: address.Address to_address;
+    //2: address.Address from_address;
+    1: Address to_address;
+    2: Address from_address;
+    3: i64 request_id;
+    4: HighLevelUnion high_level_union;
+}
+
+service UberServer {
+    oneway void ReceiveUberMessage(1: UberMessage uber_message)
+}

src/io/CMakeLists.txt

@@ -1,3 +1,5 @@
+include(MgThrift)
+
 set(io_src_files
     network/addrinfo.cpp
     network/endpoint.cpp
@@ -8,4 +10,4 @@ find_package(fmt REQUIRED)
 find_package(Threads REQUIRED)
 
 add_library(mg-io STATIC ${io_src_files})
-target_link_libraries(mg-io stdc++fs Threads::Threads fmt::fmt mg-utils)
+target_link_libraries(mg-io stdc++fs Threads::Threads fmt::fmt mg-utils FBThrift::thriftcpp2 mg-interface-ubermessage-cpp2)

src/io/message_conversion.hpp

@@ -13,14 +13,32 @@
 
 #include "io/transport.hpp"
 
-namespace memgraph::io::simulator {
+namespace memgraph::io {
 
-using memgraph::io::Duration;
-using memgraph::io::Message;
-using memgraph::io::Time;
+struct PromiseKey {
+  Address requester_address;
+  uint64_t request_id;
+  // TODO(tyler) possibly remove replier_address from promise key
+  // once we want to support DSR.
+  Address replier_address;
+
+ public:
+  friend bool operator<(const PromiseKey &lhs, const PromiseKey &rhs) {
+    if (lhs.requester_address != rhs.requester_address) {
+      return lhs.requester_address < rhs.requester_address;
+    }
+
+    if (lhs.request_id != rhs.request_id) {
+      return lhs.request_id < rhs.request_id;
+    }
+
+    return lhs.replier_address < rhs.replier_address;
+  }
+};
 
 struct OpaqueMessage {
   Address from_address;
+  Address to_address;
   uint64_t request_id;
   std::any message;
 
@@ -174,4 +192,9 @@ class OpaquePromise {
   }
 };
 
-}  // namespace memgraph::io::simulator
+struct DeadlineAndOpaquePromise {
+  Time deadline;
+  OpaquePromise promise;
+};
+
+}  // namespace memgraph::io

src/io/rsm/coordinator_rsm.hpp

@@ -0,0 +1,25 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#pragma once
+
+#include "coordinator/coordinator.hpp"
+#include "io/rsm/raft.hpp"
+
+namespace memgraph::io::rsm {
+
+// TODO(tyler) don't
+using namespace memgraph::coordinator;
+
+template <typename IoImpl>
+using CoordinatorRsm = io::rsm::Raft<IoImpl, Coordinator, WriteRequests, WriteResponses, ReadRequests, ReadResponses>;
+
+}  // namespace memgraph::io::rsm

src/io/rsm/raft.hpp

@@ -0,0 +1,844 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+// TODO(tyler) buffer out-of-order Append buffers on the Followers to reassemble more quickly
+// TODO(tyler) handle granular batch sizes based on simple flow control
+
+#pragma once
+
+#include <deque>
+#include <iostream>
+#include <map>
+#include <set>
+#include <thread>
+#include <unordered_map>
+#include <vector>
+
+#include "io/simulator/simulator.hpp"
+#include "io/transport.hpp"
+
+namespace memgraph::io::rsm {
+
+using memgraph::io::Address;
+using memgraph::io::Duration;
+using memgraph::io::Io;
+using memgraph::io::ResponseFuture;
+using memgraph::io::ResponseResult;
+using memgraph::io::Time;
+
+using Term = uint64_t;
+using LogIndex = uint64_t;
+using RequestId = uint64_t;
+
+template <typename WriteOperation>
+struct WriteRequest {
+  WriteOperation operation;
+};
+
+/// WriteResponse is returned to a client after
+/// their WriteRequest was entered in to the raft
+/// log and it reached consensus.
+///
+/// WriteReturn is the result of applying the WriteRequest to
+/// ReplicatedState, and if the ReplicatedState::write
+/// method is deterministic, all replicas will
+/// have the same ReplicatedState after applying
+/// the submitted WriteRequest.
+template <typename WriteReturn>
+struct WriteResponse {
+  bool success;
+  WriteReturn write_return;
+  std::optional<Address> retry_leader;
+};
+
+template <typename ReadOperation>
+struct ReadRequest {
+  ReadOperation operation;
+};
+
+template <typename ReadReturn>
+struct ReadResponse {
+  bool success;
+  ReadReturn read_return;
+  std::optional<Address> retry_leader;
+};
+
+/// AppendRequest is a raft-level message that the Leader
+/// periodically broadcasts to all Follower peers. This
+/// serves three main roles:
+/// 1. acts as a heartbeat from the Leader to the Follower
+/// 2. replicates new data that the Leader has received to the Follower
+/// 3. informs Follower peers when the commit index has increased,
+///    signalling that it is now safe to apply log items to the
+///    replicated state machine
+template <typename WriteRequest>
+struct AppendRequest {
+  Term term = 0;
+  LogIndex last_log_index;
+  Term last_log_term;
+  std::vector<std::pair<Term, WriteRequest>> entries;
+  LogIndex leader_commit;
+};
+
+struct AppendResponse {
+  bool success;
+  Term term;
+  Term last_log_term;
+  // a small optimization over the raft paper, tells
+  // the leader the offset that we are interested in
+  // to send log offsets from for us. This will only
+  // be useful at the beginning of a leader's term.
+  LogIndex last_log_index;
+};
+
+struct VoteRequest {
+  Term term = 0;
+  LogIndex last_log_index;
+  Term last_log_term;
+};
+
+struct VoteResponse {
+  Term term = 0;
+  LogIndex committed_log_size;
+  bool vote_granted = false;
+};
+
+template <typename WriteRequest>
+struct CommonState {
+  Term term = 0;
+  std::vector<std::pair<Term, WriteRequest>> log;
+  LogIndex committed_log_size = 0;
+  LogIndex applied_size = 0;
+};
+
+struct FollowerTracker {
+  LogIndex next_index = 0;
+  LogIndex confirmed_contiguous_index = 0;
+};
+
+struct PendingClientRequest {
+  RequestId request_id;
+  Address address;
+  Time received_at;
+};
+
+struct Leader {
+  std::map<Address, FollowerTracker> followers;
+  std::unordered_map<LogIndex, PendingClientRequest> pending_client_requests;
+  Time last_broadcast = Time::min();
+
+  std::string ToString() { return "\tLeader   \t"; }
+};
+
+struct Candidate {
+  std::map<Address, LogIndex> successful_votes;
+  Time election_began = Time::min();
+  std::set<Address> outstanding_votes;
+
+  std::string ToString() { return "\tCandidate\t"; }
+};
+
+struct Follower {
+  Time last_received_append_entries_timestamp;
+  Address leader_address;
+
+  std::string ToString() { return "\tFollower \t"; }
+};
+
+using Role = std::variant<Candidate, Leader, Follower>;
+
+/*
+all ReplicatedState classes should have an Apply method
+that returns our WriteResponseValue after consensus, and
+a Read method that returns our ReadResponseValue without
+requiring consensus.
+
+ReadResponse Read(ReadOperation);
+WriteResponseValue ReplicatedState::Apply(WriteRequest);
+
+For example:
+If the state is uint64_t, and WriteRequest is `struct PlusOne {};`,
+and WriteResponseValue is also uint64_t (the new value), then
+each call to state.Apply(PlusOne{}) will return the new value
+after incrementing it. 0, 1, 2, 3... and this will be sent back
+to the client that requested the mutation.
+
+In practice, these mutations will usually be predicated on some
+previous value, so that they are idempotent, functioning similarly
+to a CAS operation.
+*/
+template <typename WriteOperation, typename ReadOperation, typename ReplicatedState, typename WriteResponseValue,
+          typename ReadResponseValue>
+concept Rsm = requires(ReplicatedState state, WriteOperation w, ReadOperation r) {
+  { state.Read(r) } -> std::same_as<ReadResponseValue>;
+  { state.Apply(w) } -> std::same_as<WriteResponseValue>;
+};
+
+/// Parameter           Purpose
+/// --------------------------
+/// IoImpl              the concrete Io provider - SimulatorTransport, ThriftTransport, etc...
+/// ReplicatedState     the high-level data structure that is managed by the raft-backed replicated state machine
+/// WriteOperation      the individual operation type that is applied to the ReplicatedState in identical order
+///                     across each replica
+/// WriteResponseValue  the return value of calling ReplicatedState::Apply(WriteOperation), which is executed in
+///                     identical order across all replicas after an WriteOperation reaches consensus.
+/// ReadOperation       the type of operations that do not require consensus before executing directly
+///                     on a const ReplicatedState &
+/// ReadResponseValue   the return value of calling ReplicatedState::Read(ReadOperation), which is executed directly
+///                     without going through consensus first
+template <typename IoImpl, typename ReplicatedState, typename WriteOperation, typename WriteResponseValue,
+          typename ReadOperation, typename ReadResponseValue>
+requires Rsm<WriteOperation, ReadOperation, ReplicatedState, WriteResponseValue, ReadResponseValue>
+class Raft {
+  CommonState<WriteOperation> state_;
+  Role role_ = Candidate{};
+  Io<IoImpl> io_;
+  std::vector<Address> peers_;
+  ReplicatedState replicated_state_;
+
+ public:
+  Raft(Io<IoImpl> &&io, std::vector<Address> peers, ReplicatedState &&replicated_state)
+      : io_(std::move(io)), peers_(peers), replicated_state_(std::move(replicated_state)) {}
+
+  void Run() {
+    Time last_cron = io_.Now();
+
+    while (!io_.ShouldShutDown()) {
+      const auto now = io_.Now();
+      const Duration random_cron_interval = RandomTimeout(1000, 2000);
+      if (now - last_cron > random_cron_interval) {
+        Cron();
+        last_cron = now;
+      }
+
+      Duration receive_timeout = RandomTimeout(10000, 50000);
+
+      auto request_result =
+          io_.template ReceiveWithTimeout<ReadRequest<ReadOperation>, AppendRequest<WriteOperation>, AppendResponse,
+                                          WriteRequest<WriteOperation>, VoteRequest, VoteResponse>(receive_timeout);
+      if (request_result.HasError()) {
+        continue;
+      }
+
+      auto request = std::move(request_result.GetValue());
+
+      Handle(std::move(request.message), request.request_id, request.from_address);
+    }
+  }
+
+ private:
+  // Raft paper - 5.3
+  // When the entry has been safely replicated, the leader applies the
+  // entry to its state machine and returns the result of that
+  // execution to the client.
+  //
+  // "Safely replicated" is defined as being known to be present
+  // on at least a majority of all peers (inclusive of the Leader).
+  void BumpCommitIndexAndReplyToClients(Leader &leader) {
+    auto indices = std::vector<LogIndex>{};
+
+    // We include our own log size in the calculation of the log
+    // index that is present on at least a majority of all peers.
+    indices.push_back(state_.log.size());
+
+    for (const auto &[addr, f] : leader.followers) {
+      indices.push_back(f.confirmed_contiguous_index);
+      Log("at port ", addr.last_known_port, " has confirmed contiguous index of: ", f.confirmed_contiguous_index);
+    }
+
+    // reverse sort from highest to lowest (using std::ranges::greater)
+    std::ranges::sort(indices, std::ranges::greater());
+
+    // This is a particularly correctness-critical calculation because it
+    // determines which index we will consider to be the committed index.
+    //
+    // If the following indexes are recorded for clusters of different sizes,
+    // these are the expected indexes that are considered to have reached
+    // consensus:
+    // state           | expected value | (indices.size() / 2)
+    // [1]              1                 (1 / 2) => 0
+    // [2, 1]           1                 (2 / 2) => 1
+    // [3, 2, 1]        2                 (3 / 2) => 1
+    // [4, 3, 2, 1]     2                 (4 / 2) => 2
+    // [5, 4, 3, 2, 1]  3                 (5 / 2) => 2
+    size_t index_present_on_majority = indices.size() / 2;
+    LogIndex new_committed_log_size = indices[index_present_on_majority];
+
+    // We never go backwards in history.
+    MG_ASSERT(state_.committed_log_size <= new_committed_log_size);
+
+    state_.committed_log_size = new_committed_log_size;
+
+    // For each index between the old index and the new one (inclusive),
+    // Apply that log's WriteOperation to our replicated_state_,
+    // and use the specific return value of the ReplicatedState::Apply
+    // method (WriteResponseValue) to respond to the requester.
+    for (; state_.applied_size < state_.committed_log_size; state_.applied_size++) {
+      const LogIndex apply_index = state_.applied_size;
+      const auto &write_request = state_.log[apply_index].second;
+      WriteResponseValue write_return = replicated_state_.Apply(write_request);
+
+      if (leader.pending_client_requests.contains(apply_index)) {
+        PendingClientRequest client_request = std::move(leader.pending_client_requests.at(apply_index));
+
+        WriteResponse<WriteResponseValue> resp;
+        resp.success = true;
+        resp.write_return = std::move(write_return);
+
+        io_.Send(client_request.address, client_request.request_id, std::move(resp));
+        leader.pending_client_requests.erase(apply_index);
+      }
+    }
+
+    Log("committed_log_size is now ", state_.committed_log_size);
+  }
+
+  // Raft paper - 5.1
+  // AppendEntries RPCs are initiated by leaders to replicate log entries and to provide a form of heartbeat
+  void BroadcastAppendEntries(std::map<Address, FollowerTracker> &followers) {
+    for (auto &[address, follower] : followers) {
+      const LogIndex index = follower.confirmed_contiguous_index;
+
+      std::vector<std::pair<Term, WriteOperation>> entries;
+
+      if (state_.log.size() > index) {
+        entries.insert(entries.begin(), state_.log.begin() + index, state_.log.end());
+      }
+
+      const Term previous_term_from_index = PreviousTermFromIndex(index);
+
+      Log("sending ", entries.size(), " entries to Follower ", address.last_known_port,
+          " which are above its known index of ", index);
+
+      AppendRequest<WriteOperation> ar{
+          .term = state_.term,
+          .last_log_index = index,
+          .last_log_term = previous_term_from_index,
+          .entries = entries,
+          .leader_commit = state_.committed_log_size,
+      };
+
+      // request_id not necessary to set because it's not a Future-backed Request.
+      static constexpr RequestId request_id = 0;
+
+      io_.Send(address, request_id, ar);
+    }
+  }
+
+  // Raft paper - 5.2
+  // Raft uses randomized election timeouts to ensure that split votes are rare and that they are resolved quickly
+  Duration RandomTimeout(Duration min, Duration max) {
+    std::uniform_int_distribution time_distrib(min.count(), max.count());
+
+    auto rand_micros = io_.Rand(time_distrib);
+
+    return std::chrono::microseconds{rand_micros};
+  }
+
+  Duration RandomTimeout(int min_micros, int max_micros) {
+    std::uniform_int_distribution time_distrib(min_micros, max_micros);
+
+    int rand_micros = io_.Rand(time_distrib);
+
+    return std::chrono::microseconds{rand_micros};
+  }
+
+  Term PreviousTermFromIndex(LogIndex index) const {
+    if (index == 0 || state_.log.size() + 1 <= index) {
+      return 0;
+    }
+
+    const auto &[term, data] = state_.log.at(index - 1);
+    return term;
+  }
+
+  LogIndex CommittedLogIndex() { return state_.committed_log_size; }
+
+  Term CommittedLogTerm() {
+    MG_ASSERT(state_.log.size() >= state_.committed_log_size);
+    if (state_.log.empty() || state_.committed_log_size == 0) {
+      return 0;
+    }
+
+    auto &[term, data] = state_.log.at(state_.committed_log_size - 1);
+    return term;
+  }
+
+  LogIndex LastLogIndex() { return state_.log.size(); }
+
+  Term LastLogTerm() const {
+    if (state_.log.empty()) {
+      return 0;
+    }
+
+    const auto &[term, data] = state_.log.back();
+    return term;
+  }
+
+  template <typename... Ts>
+  void Log(Ts &&...args) {
+    const Time now = io_.Now();
+    auto micros = std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count();
+
+    const Term term = state_.term;
+
+    std::ostringstream out;
+
+    out << '\t' << (int)micros << "\t" << term << "\t" << io_.GetAddress().last_known_port;
+
+    std::string role_string = std::visit([&](auto &&role) { return role.ToString(); }, role_);
+
+    out << role_string;
+
+    (out << ... << args);
+
+    spdlog::debug(out.str());
+  }
+
+  /////////////////////////////////////////////////////////////
+  /// Raft-related Cron methods
+  ///
+  /// Cron + std::visit is how events are dispatched
+  /// to certain code based on Raft role.
+  ///
+  /// Cron(role) takes as the first argument a reference to its
+  /// role, and as the second argument, the message that has
+  /// been received.
+  /////////////////////////////////////////////////////////////
+
+  /// Periodic protocol maintenance.
+  void Cron() {
+    // dispatch periodic logic based on our role to a specific Cron method.
+    std::optional<Role> new_role = std::visit([&](auto &&role) { return Cron(role); }, role_);
+
+    if (new_role) {
+      role_ = std::move(new_role).value();
+    }
+  }
+
+  // Raft paper - 5.2
+  // Candidates keep sending Vote to peers until:
+  // 1. receiving Append with a higher term (become Follower)
+  // 2. receiving Vote with a higher term (become a Follower)
+  // 3. receiving a quorum of responses to our last batch of Vote (become a Leader)
+  std::optional<Role> Cron(Candidate &candidate) {
+    const auto now = io_.Now();
+    const Duration election_timeout = RandomTimeout(100000, 200000);
+    const auto election_timeout_us = std::chrono::duration_cast<std::chrono::milliseconds>(election_timeout).count();
+
+    if (now - candidate.election_began > election_timeout) {
+      state_.term++;
+      Log("becoming Candidate for term ", state_.term, " after leader timeout of ", election_timeout_us,
+          " elapsed since last election attempt");
+
+      const VoteRequest request{
+          .term = state_.term,
+          .last_log_index = LastLogIndex(),
+          .last_log_term = LastLogTerm(),
+      };
+
+      auto outstanding_votes = std::set<Address>();
+
+      for (const auto &peer : peers_) {
+        // request_id not necessary to set because it's not a Future-backed Request.
+        static constexpr auto request_id = 0;
+        io_.template Send<VoteRequest>(peer, request_id, request);
+        outstanding_votes.insert(peer);
+      }
+
+      return Candidate{
+          .successful_votes = std::map<Address, LogIndex>(),
+          .election_began = now,
+          .outstanding_votes = outstanding_votes,
+      };
+    }
+    return std::nullopt;
+  }
+
+  // Raft paper - 5.2
+  // Followers become candidates if we haven't heard from the leader
+  // after a randomized timeout.
+  std::optional<Role> Cron(Follower &follower) {
+    const auto now = io_.Now();
+    const auto time_since_last_append_entries = now - follower.last_received_append_entries_timestamp;
+    Duration election_timeout = RandomTimeout(100000, 200000);
+
+    // randomized follower timeout with a range of 100-150ms.
+    if (time_since_last_append_entries > election_timeout) {
+      // become a Candidate if we haven't heard from the Leader after this timeout
+      return Candidate{};
+    }
+
+    return std::nullopt;
+  }
+
+  // Leaders (re)send AppendRequest to followers.
+  std::optional<Role> Cron(Leader &leader) {
+    const Time now = io_.Now();
+    const Duration broadcast_timeout = RandomTimeout(40000, 60000);
+
+    if (now - leader.last_broadcast > broadcast_timeout) {
+      BroadcastAppendEntries(leader.followers);
+      leader.last_broadcast = now;
+    }
+
+    return std::nullopt;
+  }
+
+  /////////////////////////////////////////////////////////////
+  /// Raft-related Handle methods
+  ///
+  /// Handle + std::visit is how events are dispatched
+  /// to certain code based on Raft role.
+  ///
+  /// Handle(role, message, ...)
+  /// takes as the first argument a reference
+  /// to its role, and as the second argument, the
+  /// message that has been received.
+  /////////////////////////////////////////////////////////////
+
+  void Handle(std::variant<ReadRequest<ReadOperation>, AppendRequest<WriteOperation>, AppendResponse,
+                           WriteRequest<WriteOperation>, VoteRequest, VoteResponse> &&message_variant,
+              RequestId request_id, Address from_address) {
+    // dispatch the message to a handler based on our role,
+    // which can be specified in the Handle first argument,
+    // or it can be `auto` if it's a handler for several roles
+    // or messages.
+    std::optional<Role> new_role =
+        std::visit([&](auto &&msg, auto &&role) { return Handle(role, std::move(msg), request_id, from_address); },
+                   std::move(message_variant), role_);
+
+    // TODO(tyler) (M3) maybe replace std::visit with get_if for explicit prioritized matching, [[likely]] etc...
+    if (new_role) {
+      role_ = std::move(new_role).value();
+    }
+  }
+
+  // all roles can receive Vote and possibly become a follower
+  template <typename AllRoles>
+  std::optional<Role> Handle(AllRoles &, VoteRequest &&req, RequestId request_id, Address from_address) {
+    Log("received Vote from ", from_address.last_known_port, " with term ", req.term);
+    const bool last_log_term_dominates = req.last_log_term >= LastLogTerm();
+    const bool term_dominates = req.term > state_.term;
+    const bool last_log_index_dominates = req.last_log_index >= LastLogIndex();
+    const bool new_leader = last_log_term_dominates && term_dominates && last_log_index_dominates;
+
+    if (new_leader) {
+      MG_ASSERT(req.term > state_.term);
+      MG_ASSERT(std::max(req.term, state_.term) == req.term);
+    }
+
+    const VoteResponse res{
+        .term = std::max(req.term, state_.term),
+        .committed_log_size = state_.committed_log_size,
+        .vote_granted = new_leader,
+    };
+
+    io_.Send(from_address, request_id, res);
+
+    if (new_leader) {
+      // become a follower
+      state_.term = req.term;
+      return Follower{
+          .last_received_append_entries_timestamp = io_.Now(),
+          .leader_address = from_address,
+      };
+    } else if (term_dominates) {
+      Log("received a vote from an inferior candidate. Becoming Candidate");
+      state_.term = std::max(state_.term, req.term) + 1;
+      return Candidate{};
+    }
+
+    return std::nullopt;
+  }
+
+  std::optional<Role> Handle(Candidate &candidate, VoteResponse &&res, RequestId, Address from_address) {
+    Log("received VoteResponse");
+
+    if (!res.vote_granted || res.term != state_.term) {
+      Log("received unsuccessful VoteResponse from term ", res.term, " when our candidacy term is ", state_.term);
+      // we received a delayed VoteResponse from the past, which has to do with an election that is
+      // no longer valid. We can simply drop this.
+      return std::nullopt;
+    }
+
+    MG_ASSERT(candidate.outstanding_votes.contains(from_address),
+              "Received unexpected VoteResponse from server not present in Candidate's outstanding_votes!");
+    candidate.outstanding_votes.erase(from_address);
+
+    MG_ASSERT(!candidate.successful_votes.contains(from_address),
+              "Received unexpected VoteResponse from server already in Candidate's successful_votes!");
+    candidate.successful_votes.insert({from_address, res.committed_log_size});
+
+    if (candidate.successful_votes.size() >= candidate.outstanding_votes.size()) {
+      std::map<Address, FollowerTracker> followers{};
+
+      for (const auto &[address, committed_log_size] : candidate.successful_votes) {
+        FollowerTracker follower{
+            .next_index = committed_log_size,
+            .confirmed_contiguous_index = committed_log_size,
+        };
+        followers.insert({address, follower});
+      }
+      for (const auto &address : candidate.outstanding_votes) {
+        FollowerTracker follower{
+            .next_index = state_.log.size(),
+            .confirmed_contiguous_index = 0,
+        };
+        followers.insert({address, follower});
+      }
+
+      Log("becoming Leader at term ", state_.term);
+
+      BroadcastAppendEntries(followers);
+
+      return Leader{
+          .followers = std::move(followers),
+          .pending_client_requests = std::unordered_map<LogIndex, PendingClientRequest>(),
+      };
+    }
+
+    return std::nullopt;
+  }
+
+  template <typename AllRoles>
+  std::optional<Role> Handle(AllRoles &, VoteResponse &&, RequestId, Address) {
+    Log("non-Candidate received VoteResponse");
+    return std::nullopt;
+  }
+
+  template <typename AllRoles>
+  std::optional<Role> Handle(AllRoles &role, AppendRequest<WriteOperation> &&req, RequestId request_id,
+                             Address from_address) {
+    AppendResponse res{
+        .success = false,
+        .term = state_.term,
+        .last_log_term = CommittedLogTerm(),
+        .last_log_index = CommittedLogIndex(),
+    };
+
+    if constexpr (std::is_same<AllRoles, Leader>()) {
+      MG_ASSERT(req.term != state_.term, "Multiple leaders are acting under the term ", req.term);
+    }
+
+    const bool is_candidate = std::is_same<AllRoles, Candidate>();
+    const bool is_failed_competitor = is_candidate && req.term == state_.term;
+    const Time now = io_.Now();
+
+    // Raft paper - 5.2
+    // While waiting for votes, a candidate may receive an
+    // AppendEntries RPC from another server claiming to be leader. If
+    // the leader’s term (included in its RPC) is at least as large as
+    // the candidate’s current term, then the candidate recognizes the
+    // leader as legitimate and returns to follower state.
+    if (req.term > state_.term || is_failed_competitor) {
+      // become follower of this leader, reply with our log status
+      state_.term = req.term;
+
+      io_.Send(from_address, request_id, res);
+
+      Log("becoming Follower of Leader ", from_address.last_known_port, " at term ", req.term);
+      return Follower{
+          .last_received_append_entries_timestamp = now,
+          .leader_address = from_address,
+      };
+    } else if (req.term < state_.term) {
+      // nack this request from an old leader
+      io_.Send(from_address, request_id, res);
+
+      return std::nullopt;
+    }
+
+    // at this point, we're dealing with our own leader
+    if constexpr (std::is_same<AllRoles, Follower>()) {
+      // small specialization for when we're already a Follower
+      MG_ASSERT(role.leader_address == from_address, "Multiple Leaders are acting under the same term number!");
+      role.last_received_append_entries_timestamp = now;
+    } else {
+      Log("Somehow entered Follower-specific logic as a non-Follower");
+      MG_ASSERT(false, "Somehow entered Follower-specific logic as a non-Follower");
+    }
+
+    res.last_log_term = LastLogTerm();
+    res.last_log_index = LastLogIndex();
+
+    Log("returning last_log_index of ", res.last_log_index);
+
+    // Handle steady-state conditions.
+    if (req.last_log_index != LastLogIndex()) {
+      Log("req.last_log_index is above our last applied log index");
+    } else if (req.last_log_term != LastLogTerm()) {
+      Log("req.last_log_term differs from our leader term at that slot, expected: ", LastLogTerm(), " but got ",
+          req.last_log_term);
+    } else {
+      // happy path - Apply log
+      Log("applying batch of entries to log of size ", req.entries.size());
+
+      MG_ASSERT(req.last_log_index >= state_.committed_log_size,
+                "Applied history from Leader which goes back in time from our commit_index");
+
+      // possibly chop-off stuff that was replaced by
+      // things with different terms (we got data that
+      // hasn't reached consensus yet, which is normal)
+      state_.log.resize(req.last_log_index);
+
+      state_.log.insert(state_.log.end(), req.entries.begin(), req.entries.end());
+
+      MG_ASSERT(req.leader_commit >= state_.committed_log_size);
+      state_.committed_log_size = std::min(req.leader_commit, LastLogIndex());
+
+      for (; state_.applied_size < state_.committed_log_size; state_.applied_size++) {
+        const auto &write_request = state_.log[state_.applied_size].second;
+        replicated_state_.Apply(write_request);
+      }
+
+      res.success = true;
+    }
+
+    io_.Send(from_address, request_id, res);
+
+    return std::nullopt;
+  }
+
+  std::optional<Role> Handle(Leader &leader, AppendResponse &&res, RequestId, Address from_address) {
+    if (res.term != state_.term) {
+    } else if (!leader.followers.contains(from_address)) {
+      Log("received AppendResponse from unknown Follower");
+      MG_ASSERT(false, "received AppendResponse from unknown Follower");
+    } else {
+      if (res.success) {
+        Log("got successful AppendResponse from ", from_address.last_known_port, " with last_log_index of ",
+            res.last_log_index);
+      } else {
+        Log("got unsuccessful AppendResponse from ", from_address.last_known_port, " with last_log_index of ",
+            res.last_log_index);
+      }
+      FollowerTracker &follower = leader.followers.at(from_address);
+      follower.next_index = std::max(follower.next_index, res.last_log_index);
+      follower.confirmed_contiguous_index = std::max(follower.confirmed_contiguous_index, res.last_log_index);
+
+      BumpCommitIndexAndReplyToClients(leader);
+    }
+    return std::nullopt;
+  }
+
+  template <typename AllRoles>
+  std::optional<Role> Handle(AllRoles &, AppendResponse &&, RequestId, Address) {
+    // we used to be the leader, and are getting old delayed responses
+    return std::nullopt;
+  }
+
+  /////////////////////////////////////////////////////////////
+  /// RSM-related handle methods
+  /////////////////////////////////////////////////////////////
+
+  // Leaders are able to immediately respond to the requester (with a ReadResponseValue) applied to the ReplicatedState
+  std::optional<Role> Handle(Leader &, ReadRequest<ReadOperation> &&req, RequestId request_id, Address from_address) {
+    Log("handling ReadOperation");
+    ReadOperation read_operation = req.operation;
+
+    ReadResponseValue read_return = replicated_state_.Read(read_operation);
+
+    ReadResponse<ReadResponseValue> resp{
+        .success = true,
+        .read_return = std::move(read_return),
+        .retry_leader = std::nullopt,
+    };
+
+    io_.Send(from_address, request_id, resp);
+
+    return std::nullopt;
+  }
+
+  // Candidates should respond with a failure, similar to the Candidate + WriteRequest failure below
+  std::optional<Role> Handle(Candidate &, ReadRequest<ReadOperation> &&, RequestId request_id, Address from_address) {
+    Log("received ReadOperation - not redirecting because no Leader is known");
+    auto res = ReadResponse<ReadResponseValue>{};
+
+    res.success = false;
+
+    Cron();
+
+    io_.Send(from_address, request_id, res);
+
+    return std::nullopt;
+  }
+
+  // Followers should respond with a redirection, similar to the Follower + WriteRequest response below
+  std::optional<Role> Handle(Follower &follower, ReadRequest<ReadOperation> &&, RequestId request_id,
+                             Address from_address) {
+    auto res = ReadResponse<ReadResponseValue>{};
+
+    res.success = false;
+    Log("redirecting client to known Leader with port ", follower.leader_address.last_known_port);
+    res.retry_leader = follower.leader_address;
+
+    io_.Send(from_address, request_id, res);
+
+    return std::nullopt;
+  }
+
+  // Raft paper - 8
+  // When a client first starts up, it connects to a randomly chosen
+  // server. If the client’s first choice is not the leader, that
+  // server will reject the client’s request and supply information
+  // about the most recent leader it has heard from.
+  std::optional<Role> Handle(Follower &follower, WriteRequest<WriteOperation> &&, RequestId request_id,
+                             Address from_address) {
+    auto res = WriteResponse<WriteResponseValue>{};
+
+    res.success = false;
+    Log("redirecting client to known Leader with port ", follower.leader_address.last_known_port);
+    res.retry_leader = follower.leader_address;
+
+    io_.Send(from_address, request_id, res);
+
+    return std::nullopt;
+  }
+
+  std::optional<Role> Handle(Candidate &, WriteRequest<WriteOperation> &&, RequestId request_id, Address from_address) {
+    Log("received WriteRequest - not redirecting because no Leader is known");
+    auto res = WriteResponse<WriteResponseValue>{};
+
+    res.success = false;
+
+    Cron();
+
+    io_.Send(from_address, request_id, res);
+
+    return std::nullopt;
+  }
+
+  // only leaders actually handle replication requests from clients
+  std::optional<Role> Handle(Leader &leader, WriteRequest<WriteOperation> &&req, RequestId request_id,
+                             Address from_address) {
+    Log("handling WriteRequest");
+
+    // we are the leader. add item to log and send Append to peers
+    state_.log.emplace_back(std::pair(state_.term, std::move(req.operation)));
+
+    LogIndex log_index = state_.log.size() - 1;
+
+    PendingClientRequest pcr{
+        .request_id = request_id,
+        .address = from_address,
+        .received_at = io_.Now(),
+    };
+
+    leader.pending_client_requests.emplace(log_index, pcr);
+
+    BroadcastAppendEntries(leader.followers);
+
+    return std::nullopt;
+  }
+};
+
+};  // namespace memgraph::io::rsm

src/io/rsm/rsm_client.hpp

@@ -0,0 +1,137 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#include <iostream>
+#include <optional>
+#include <vector>
+
+#include "io/address.hpp"
+#include "io/rsm/raft.hpp"
+#include "utils/result.hpp"
+
+namespace memgraph::io::rsm {
+
+using memgraph::io::Address;
+using memgraph::io::Duration;
+using memgraph::io::ResponseEnvelope;
+using memgraph::io::ResponseFuture;
+using memgraph::io::ResponseResult;
+using memgraph::io::Time;
+using memgraph::io::TimedOut;
+using memgraph::io::rsm::ReadRequest;
+using memgraph::io::rsm::ReadResponse;
+using memgraph::io::rsm::WriteRequest;
+using memgraph::io::rsm::WriteResponse;
+using memgraph::utils::BasicResult;
+
+template <typename IoImpl, typename WriteRequestT, typename WriteResponseT, typename ReadRequestT,
+          typename ReadResponseT>
+class RsmClient {
+  using ServerPool = std::vector<Address>;
+
+  IoImpl io_;
+  Address leader_;
+
+  std::mt19937 cli_rng_{0};
+  ServerPool server_addrs_;
+
+  template <typename ResponseT>
+  void PossiblyRedirectLeader(const ResponseT &response) {
+    if (response.retry_leader) {
+      MG_ASSERT(!response.success, "retry_leader should never be set for successful responses");
+      leader_ = response.retry_leader.value();
+      spdlog::debug("client redirected to leader server {}", leader_.ToString());
+    } else if (!response.success) {
+      std::uniform_int_distribution<size_t> addr_distrib(0, (server_addrs_.size() - 1));
+      size_t addr_index = addr_distrib(cli_rng_);
+      leader_ = server_addrs_[addr_index];
+
+      spdlog::debug(
+          "client NOT redirected to leader server despite our success failing to be processed (it probably was sent to "
+          "a RSM Candidate) trying a random one at index {} with address {}",
+          addr_index, leader_.ToString());
+    }
+  }
+
+ public:
+  RsmClient(IoImpl io, Address leader, ServerPool server_addrs)
+      : io_{io}, leader_{leader}, server_addrs_{server_addrs} {}
+
+  RsmClient() = delete;
+
+  BasicResult<TimedOut, WriteResponseT> SendWriteRequest(WriteRequestT req) {
+    WriteRequest<WriteRequestT> client_req;
+    client_req.operation = req;
+
+    const Duration overall_timeout = io_.GetDefaultTimeout();
+    const Time before = io_.Now();
+
+    do {
+      spdlog::debug("client sending WriteRequest to Leader {}", leader_.ToString());
+      ResponseFuture<WriteResponse<WriteResponseT>> response_future =
+          io_.template Request<WriteRequest<WriteRequestT>, WriteResponse<WriteResponseT>>(leader_, client_req);
+      ResponseResult<WriteResponse<WriteResponseT>> response_result = std::move(response_future).Wait();
+
+      if (response_result.HasError()) {
+        spdlog::debug("client timed out while trying to communicate with leader server {}", leader_.ToString());
+        // continue;
+        return response_result.GetError();
+      }
+
+      ResponseEnvelope<WriteResponse<WriteResponseT>> &&response_envelope = std::move(response_result.GetValue());
+      WriteResponse<WriteResponseT> &&write_response = std::move(response_envelope.message);
+
+      if (write_response.success) {
+        return std::move(write_response.write_return);
+      }
+
+      PossiblyRedirectLeader(write_response);
+    } while (io_.Now() < before + overall_timeout);
+
+    return TimedOut{};
+  }
+
+  BasicResult<TimedOut, ReadResponseT> SendReadRequest(ReadRequestT req) {
+    ReadRequest<ReadRequestT> read_req;
+    read_req.operation = req;
+
+    const Duration overall_timeout = io_.GetDefaultTimeout();
+    const Time before = io_.Now();
+
+    do {
+      spdlog::debug("client sending ReadRequest to Leader {}", leader_.ToString());
+
+      ResponseFuture<ReadResponse<ReadResponseT>> get_response_future =
+          io_.template Request<ReadRequest<ReadRequestT>, ReadResponse<ReadResponseT>>(leader_, read_req);
+
+      // receive response
+      ResponseResult<ReadResponse<ReadResponseT>> get_response_result = std::move(get_response_future).Wait();
+
+      if (get_response_result.HasError()) {
+        spdlog::debug("client timed out while trying to communicate with leader server {}", leader_.ToString());
+        return get_response_result.GetError();
+      }
+
+      ResponseEnvelope<ReadResponse<ReadResponseT>> &&get_response_envelope = std::move(get_response_result.GetValue());
+      ReadResponse<ReadResponseT> &&read_get_response = std::move(get_response_envelope.message);
+
+      if (read_get_response.success) {
+        return std::move(read_get_response.read_return);
+      }
+
+      PossiblyRedirectLeader(read_get_response);
+    } while (io_.Now() < before + overall_timeout);
+
+    return TimedOut{};
+  }
+};
+
+}  // namespace memgraph::io::rsm

src/io/rsm/shard_rsm.hpp

@@ -0,0 +1,165 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#pragma once
+
+/// The StorageRsm is a simple in-memory raft-backed kv store that can be used for simple testing
+/// and implementation of some query engine logic before storage engines are fully implemented.
+///
+/// To implement multiple read and write commands, change the StorageRead* and StorageWrite* requests
+/// and responses to a std::variant of the different options, and route them to specific handlers in
+/// the StorageRsm's Read and Apply methods. Remember that Read is called immediately when the Raft
+/// leader receives the request, and does not replicate anything over Raft. Apply is called only
+/// AFTER the StorageWriteRequest is replicated to a majority of Raft peers, and the result of calling
+/// StorageRsm::Apply(StorageWriteRequest) is returned to the client that submitted the request.
+
+#include <deque>
+#include <iostream>
+#include <map>
+#include <optional>
+#include <set>
+#include <thread>
+#include <vector>
+
+#include "coordinator/hybrid_logical_clock.hpp"
+#include "io/address.hpp"
+#include "io/rsm/raft.hpp"
+#include "io/simulator/simulator.hpp"
+#include "io/simulator/simulator_transport.hpp"
+#include "storage/v2/property_value.hpp"
+#include "utils/logging.hpp"
+
+using memgraph::coordinator::Hlc;
+using memgraph::io::Address;
+using memgraph::io::Io;
+using memgraph::io::ResponseEnvelope;
+using memgraph::io::ResponseFuture;
+using memgraph::io::ResponseResult;
+using memgraph::io::rsm::Raft;
+using memgraph::io::rsm::ReadRequest;
+using memgraph::io::rsm::ReadResponse;
+using memgraph::io::rsm::WriteRequest;
+using memgraph::io::rsm::WriteResponse;
+using memgraph::io::simulator::Simulator;
+using memgraph::io::simulator::SimulatorConfig;
+using memgraph::io::simulator::SimulatorStats;
+using memgraph::io::simulator::SimulatorTransport;
+using memgraph::storage::PropertyValue;
+
+namespace memgraph::io::rsm {
+
+using ShardRsmKey = std::vector<PropertyValue>;
+
+struct StorageWriteRequest {
+  ShardRsmKey key;
+  std::optional<int> value;
+};
+
+struct StorageWriteResponse {
+  bool shard_rsm_success;
+  std::optional<int> last_value;
+  // Only has a value if the given shard does not contain the requested key
+  std::optional<Hlc> latest_known_shard_map_version{std::nullopt};
+};
+
+struct StorageReadRequest {
+  ShardRsmKey key;
+};
+
+struct StorageReadResponse {
+  bool shard_rsm_success;
+  std::optional<int> value;
+  // Only has a value if the given shard does not contain the requested key
+  std::optional<Hlc> latest_known_shard_map_version{std::nullopt};
+};
+
+class StorageRsm {
+  std::map<ShardRsmKey, int> state_;
+  ShardRsmKey minimum_key_;
+  std::optional<ShardRsmKey> maximum_key_{std::nullopt};
+  Hlc shard_map_version_;
+
+  // The key is not located in this shard
+  bool IsKeyInRange(const ShardRsmKey &key) {
+    if (maximum_key_) [[likely]] {
+      return (key >= minimum_key_ && key <= maximum_key_);
+    }
+    return key >= minimum_key_;
+  }
+
+ public:
+  StorageReadResponse Read(StorageReadRequest request) {
+    StorageReadResponse ret;
+
+    if (!IsKeyInRange(request.key)) {
+      ret.latest_known_shard_map_version = shard_map_version_;
+      ret.shard_rsm_success = false;
+    } else if (state_.contains(request.key)) {
+      ret.value = state_[request.key];
+      ret.shard_rsm_success = true;
+    } else {
+      ret.shard_rsm_success = false;
+      ret.value = std::nullopt;
+    }
+    return ret;
+  }
+
+  StorageWriteResponse Apply(StorageWriteRequest request) {
+    StorageWriteResponse ret;
+
+    // Key is outside the prohibited range
+    if (!IsKeyInRange(request.key)) {
+      ret.latest_known_shard_map_version = shard_map_version_;
+      ret.shard_rsm_success = false;
+    }
+    // Key exist
+    else if (state_.contains(request.key)) {
+      auto &val = state_[request.key];
+
+      /*
+       *   Delete
+       */
+      if (!request.value) {
+        ret.shard_rsm_success = true;
+        ret.last_value = val;
+        state_.erase(state_.find(request.key));
+      }
+
+      /*
+       *   Update
+       */
+      // Does old_value match?
+      if (request.value == val) {
+        ret.last_value = val;
+        ret.shard_rsm_success = true;
+
+        val = request.value.value();
+
+      } else {
+        ret.last_value = val;
+        ret.shard_rsm_success = false;
+      }
+    }
+    /*
+     *   Create
+     */
+    else {
+      ret.last_value = std::nullopt;
+      ret.shard_rsm_success = true;
+
+      state_.emplace(request.key, std::move(request.value).value());
+    }
+
+    return ret;
+  }
+};
+
+}  // namespace memgraph::io::rsm

src/io/simulator/simulator.hpp

@@ -23,6 +23,7 @@ namespace memgraph::io::simulator {
 class Simulator {
   std::mt19937 rng_;
   std::shared_ptr<SimulatorHandle> simulator_handle_;
+  uint16_t auto_port_ = 0;
 
  public:
   explicit Simulator(SimulatorConfig config)
@@ -30,6 +31,11 @@ class Simulator {
 
   void ShutDown() { simulator_handle_->ShutDown(); }
 
+  Io<SimulatorTransport> RegisterNew() {
+    Address address = Address::TestAddress(auto_port_++);
+    return Register(address);
+  }
+
   Io<SimulatorTransport> Register(Address address) {
     std::uniform_int_distribution<uint64_t> seed_distrib;
     uint64_t seed = seed_distrib(rng_);

src/io/simulator/simulator_handle.hpp

@@ -24,7 +24,7 @@
 
 #include "io/address.hpp"
 #include "io/errors.hpp"
-#include "io/simulator/message_conversion.hpp"
+#include "io/message_conversion.hpp"
 #include "io/simulator/simulator_config.hpp"
 #include "io/simulator/simulator_stats.hpp"
 #include "io/time.hpp"
@@ -33,33 +33,10 @@
 namespace memgraph::io::simulator {
 
 using memgraph::io::Duration;
+using memgraph::io::OpaqueMessage;
+using memgraph::io::OpaquePromise;
 using memgraph::io::Time;
-
-struct PromiseKey {
-  Address requester_address;
-  uint64_t request_id;
-  // TODO(tyler) possibly remove replier_address from promise key
-  // once we want to support DSR.
-  Address replier_address;
-
- public:
-  friend bool operator<(const PromiseKey &lhs, const PromiseKey &rhs) {
-    if (lhs.requester_address != rhs.requester_address) {
-      return lhs.requester_address < rhs.requester_address;
-    }
-
-    if (lhs.request_id != rhs.request_id) {
-      return lhs.request_id < rhs.request_id;
-    }
-
-    return lhs.replier_address < rhs.replier_address;
-  }
-};
-
-struct DeadlineAndOpaquePromise {
-  Time deadline;
-  OpaquePromise promise;
-};
+using memgraph::io::TimedOut;
 
 class SimulatorHandle {
   mutable std::mutex mu_{};

src/io/thrift/thrift_handle.hpp

@@ -0,0 +1,249 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#pragma once
+
+#include <condition_variable>
+#include <map>
+#include <memory>
+#include <mutex>
+
+#include <boost/asio/ip/tcp.hpp>
+#include <boost/lexical_cast.hpp>
+
+#include <folly/init/Init.h>
+#include <folly/io/SocketOptionMap.h>
+#include <folly/io/async/AsyncServerSocket.h>
+#include <folly/net/NetworkSocket.h>
+#include <thrift/lib/cpp2/async/HeaderClientChannel.h>
+#include <thrift/lib/cpp2/server/ThriftServer.h>
+
+// From generated code
+#include "interface/gen-cpp2/UberServer.h"
+#include "interface/gen-cpp2/UberServerAsyncClient.h"
+
+#include "io/errors.hpp"
+#include "io/message_conversion.hpp"
+#include "io/transport.hpp"
+
+namespace memgraph::io::thrift {
+
+using namespace apache::thrift;
+using namespace folly;
+
+using memgraph::io::Address;
+using memgraph::io::OpaqueMessage;
+using memgraph::io::OpaquePromise;
+using memgraph::io::TimedOut;
+using RequestId = uint64_t;
+class ThriftHandle {
+  mutable std::mutex mu_{};
+  mutable std::condition_variable cv_;
+  const Address address_ = Address::TestAddress(0);
+
+  EventBase base_;
+
+  // the responses to requests that are being waited on
+  std::map<PromiseKey, DeadlineAndOpaquePromise> promises_;
+
+  // messages that are sent to servers that may later receive them
+  std::vector<OpaqueMessage> can_receive_;
+
+  // TODO(tyler) thrift clients for each outbound address combination
+  // AsyncClient does not offer default init so they are optional atm.
+  std::map<Address, std::optional<cpp2::UberServerAsyncClient>> clients_;
+
+  // TODO(gabor) make this to a threadpool
+  // uuid of the address -> port number where the given rsm is residing.
+  // TODO(gabor) The RSM map should not be a part of this class.
+  // std::map<boost::uuids::uuid, uint16_t /*this should be the actual RSM*/> rsm_map_;
+
+  cpp2::Address convertToUberAddress(const memgraph::io::Address &address) {
+    cpp2::Address ret_address;
+    ret_address.unique_id_ref() = boost::uuids::to_string(address.unique_id);
+    ret_address.last_known_ip_ref() = address.last_known_ip.to_string();
+    ret_address.last_known_port_ref() = static_cast<int32_t>(address.last_known_port);
+    return ret_address;
+  }
+
+ public:
+  explicit ThriftHandle(Address our_address) : address_(our_address) {}
+
+  Time Now() const {
+    auto nano_time = std::chrono::system_clock::now();
+    return std::chrono::time_point_cast<std::chrono::microseconds>(nano_time);
+  }
+
+  template <Message M>
+  void DeliverMessage(Address to_address, Address from_address, RequestId request_id, M &&message) {
+    std::any message_any(std::move(message));
+    OpaqueMessage opaque_message{
+        .from_address = from_address, .request_id = request_id, .message = std::move(message_any)};
+
+    PromiseKey promise_key{.requester_address = to_address,
+                           .request_id = opaque_message.request_id,
+                           .replier_address = opaque_message.from_address};
+
+    {
+      std::unique_lock<std::mutex> lock(mu_);
+
+      if (promises_.contains(promise_key)) {
+        // complete waiting promise if it's there
+        DeadlineAndOpaquePromise dop = std::move(promises_.at(promise_key));
+        promises_.erase(promise_key);
+
+        dop.promise.Fill(std::move(opaque_message));
+      } else {
+        can_receive_.emplace_back(std::move(opaque_message));
+      }
+    }  // lock dropped
+
+    cv_.notify_all();
+  }
+
+  template <Message Request, Message Response>
+  void SubmitRequest(Address to_address, Address from_address, RequestId request_id, Request &&request,
+                     Duration timeout, ResponsePromise<Response> &&promise) {
+    const Time deadline = Now() + timeout;
+    Address our_address = address_;
+
+    PromiseKey promise_key{.requester_address = from_address, .request_id = request_id, .replier_address = to_address};
+    OpaquePromise opaque_promise(std::move(promise).ToUnique());
+    DeadlineAndOpaquePromise dop{.deadline = deadline, .promise = std::move(opaque_promise)};
+    promises_.emplace(std::move(promise_key), std::move(dop));
+
+    cv_.notify_all();
+
+    bool port_matches = to_address.last_known_port == our_address.last_known_port;
+    bool ip_matches = to_address.last_known_ip == our_address.last_known_ip;
+
+    if (port_matches && ip_matches) {
+      // hairpin routing optimization
+      DeliverMessage(to_address, from_address, request_id, std::move(request));
+    } else {
+      // send using a thrift client to remove service
+      Send(to_address, from_address, request_id, request);
+    }
+  }
+
+  template <Message... Ms>
+  requires(sizeof...(Ms) > 0) RequestResult<Ms...> Receive(Duration timeout) {
+    // TODO(tyler) block for the specified duration on the Inbox's receipt of a message of this type.
+    std::unique_lock lock(mu_);
+
+    Time before = Now();
+
+    while (can_receive_.empty()) {
+      Time now = Now();
+
+      // protection against non-monotonic timesources
+      auto maxed_now = std::max(now, before);
+      auto elapsed = maxed_now - before;
+
+      if (timeout < elapsed) {
+        return TimedOut{};
+      }
+
+      Duration relative_timeout = timeout - elapsed;
+
+      auto cv_status_value = cv_.wait_for(lock, relative_timeout);
+
+      if (cv_status_value == std::cv_status::timeout) {
+        return TimedOut{};
+      }
+    }
+
+    auto current_message = std::move(can_receive_.back());
+    can_receive_.pop_back();
+
+    auto m_opt = std::move(current_message).Take<Ms...>();
+
+    return std::move(m_opt).value();
+  }
+
+  // This method is used for communication for in-between different
+  // machines and processes, its exact functionality will be
+  // implemented later after the shape of the Thrift generated
+  // UberMessage is specified as this is not needed for M1.
+  template <Message M>
+  void Send(Address to_address, Address from_address, RequestId request_id, M message) {
+    MG_ASSERT(false, "Communication in-between different machines and processes is not yet implemented!");
+
+    //   // TODO(tyler) call thrift client for address (or create one if it doesn't exist yet)
+    //   cpp2::UberMessage uber_message;
+
+    //   uber_message.to_address_ref() = convertToUberAddress(to_address);
+    //   uber_message.from_address_ref() = convertToUberAddress(from_address);
+    //   uber_message.request_id_ref() = static_cast<int64_t>(request_id);
+    //   uber_message.high_level_union_ref() = message;
+
+    //   // cpp2::UberMessage uber_message = {
+    //   //   .to_address = convertToUberAddress(to_address),
+    //   //   .from_address = convertToUberAddress(from_address),
+    //   //   .request_id = static_cast<int64_t>(request_id),
+    //   //   .high_level_union = message
+    //   // };
+
+    //   if (clients_.contains(to_address)) {
+    //     auto &client = clients_[to_address];
+    //     client->sync_ReceiveUberMessage(uber_message);
+    //   } else {
+    //     // maybe make this into a member var
+    //     const auto &other_ip = to_address.last_known_ip.to_string();
+    //     const auto &other_port = to_address.last_known_port;
+    //     auto socket(folly::AsyncSocket::newSocket(&base_, other_ip, other_port));
+    //     auto client_channel = HeaderClientChannel::newChannel(std::move(socket));
+    //     // Create a client object
+    //     cpp2::UberServerAsyncClient client(std::move(client_channel));
+
+    //     client.sync_ReceiveUberMessage(uber_message);
+    //   }
+  }
+};
+
+class UberMessageService final : cpp2::UberServerSvIf {
+  std::shared_ptr<ThriftHandle> handle_;
+
+  memgraph::io::Address convertToMgAddress(const cpp2::Address &address) {
+    memgraph::io::Address ret_address;
+    ret_address = {.unique_id{boost::lexical_cast<boost::uuids::uuid>(address.get_unique_id())},
+                   .last_known_ip{boost::asio::ip::make_address(address.get_last_known_ip())},
+                   .last_known_port = static_cast<uint16_t>(address.get_last_known_port())};
+    return ret_address;
+  }
+
+ public:
+  explicit UberMessageService(std::shared_ptr<ThriftHandle> handle) : handle_{handle} {}
+
+  void ReceiveUberMessage(const cpp2::UberMessage &uber_message) override {
+    const auto &to_address = uber_message.get_to_address();
+    const auto &from_address = uber_message.get_from_address();
+    const auto &request_id = uber_message.get_request_id();
+    auto message = uber_message.get_high_level_union();
+
+    const auto mg_to_address = convertToMgAddress(to_address);
+    const auto mg_from_address = convertToMgAddress(from_address);
+    // Castint int64_t -> uint64_t
+    // FBThrift only provides us with signed integers. If someone
+    // wishes to use signed integers then the go to solution seems to
+    // be to use the one-bigger signed version. Unfortunately FBThrift
+    // does not provide a uint128_t so we have to use the 64 bit one
+    // for now.
+    // TODO(gvolfing) Investigate and try to get around this problem
+    // with Varint or some other Thrift type.
+    const auto mg_request_id = static_cast<uint64_t>(request_id);
+
+    // Transform high_level_union into something usable if needed(?).
+    handle_->DeliverMessage(mg_to_address, mg_from_address, mg_request_id, std::move(message));
+  }
+};
+
+}  // namespace memgraph::io::thrift

src/io/thrift/thrift_transport.hpp

@@ -0,0 +1,66 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#pragma once
+
+#include <chrono>
+#include <memory>
+#include <random>
+#include <utility>
+
+#include "io/address.hpp"
+#include "io/thrift/thrift_handle.hpp"
+#include "io/time.hpp"
+#include "io/transport.hpp"
+
+namespace memgraph::io::thrift {
+
+using memgraph::io::Duration;
+using memgraph::io::Time;
+
+class ThriftTransport {
+  std::shared_ptr<ThriftHandle> thrift_handle_;
+  const Address address_;
+  std::random_device rng_;
+
+ public:
+  ThriftTransport(std::shared_ptr<ThriftHandle> thrift_handle, Address address)
+      : thrift_handle_(thrift_handle), address_(address) {}
+
+  template <Message Request, Message Response>
+  ResponseFuture<Response> Request(Address address, uint64_t request_id, Request request, Duration timeout) {
+    auto [future, promise] = memgraph::io::FuturePromisePair<ResponseResult<Response>>();
+
+    thrift_handle_->SubmitRequest(address, address_, request_id, std::move(request), timeout, std::move(promise));
+
+    return std::move(future);
+  }
+
+  template <Message... Ms>
+  requires(sizeof...(Ms) > 0) RequestResult<Ms...> Receive(Duration timeout) {
+    return thrift_handle_->template Receive<Ms...>(address_, timeout);
+  }
+
+  template <Message M>
+  void Send(Address address, uint64_t request_id, M message) {
+    return thrift_handle_->template Send<M>(address, address_, request_id, message);
+  }
+
+  Time Now() const { return thrift_handle_->Now(); }
+
+  bool ShouldShutDown() const { return false; }
+
+  template <class D = std::poisson_distribution<>, class Return = uint64_t>
+  Return Rand(D distrib) {
+    return distrib(rng_);
+  }
+};
+};  // namespace memgraph::io::thrift

src/io/time.hpp

@@ -16,6 +16,6 @@
 namespace memgraph::io {
 
 using Duration = std::chrono::microseconds;
-using Time = std::chrono::time_point<std::chrono::local_t, Duration>;
+using Time = std::chrono::time_point<std::chrono::system_clock, Duration>;
 
 }  // namespace memgraph::io

src/io/transport.hpp

@@ -72,6 +72,9 @@ class Io {
   /// without an explicit timeout set.
   void SetDefaultTimeout(Duration timeout) { default_timeout_ = timeout; }
 
+  /// Returns the current default timeout for this Io instance.
+  Duration GetDefaultTimeout() { return default_timeout_; }
+
   /// Issue a request with an explicit timeout in microseconds provided. This tends to be used by clients.
   template <Message Request, Message Response>
   ResponseFuture<Response> RequestWithTimeout(Address address, Request request, Duration timeout) {

src/memgraph.cpp

@@ -26,6 +26,7 @@
 #include <thread>
 
 #include <fmt/format.h>
+#include <folly/init/Init.h>
 #include <gflags/gflags.h>
 #include <spdlog/common.h>
 #include <spdlog/sinks/daily_file_sink.h>
@@ -1054,6 +1055,8 @@ int main(int argc, char **argv) {
   LoadConfig("memgraph");
   gflags::ParseCommandLineFlags(&argc, &argv, true);
 
+  folly::Init(&argc, &argv);
+
   InitializeLogger();
 
   // Unhandled exception handler init.

src/storage/v3/storage_rsm_map.hpp

@@ -0,0 +1,30 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#pragma once
+
+#include <map>
+
+#include <boost/uuid/uuid.hpp>
+#include <boost/uuid/uuid_io.hpp>
+
+#include "src/io/rsm/shard_rsm.hpp"
+
+namespace memgraph::io {
+
+using memgraph::io::rsm::StorageRsm;
+
+// TODO(gabor) make this work with a threadpool
+struct RsmMap {
+  std::map<boost::uuids::uuid, StorageRsm> map;
+};
+
+}  // namespace memgraph::io

tests/simulation/CMakeLists.txt

@@ -27,4 +27,8 @@ endfunction(add_simulation_test)
 
 add_simulation_test(basic_request.cpp address)
 
+add_simulation_test(raft.cpp address)
+
 add_simulation_test(trial_query_storage/query_storage_test.cpp address)
+
+add_simulation_test(sharded_map.cpp address)

tests/simulation/raft.cpp

@@ -0,0 +1,265 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#include <chrono>
+#include <deque>
+#include <iostream>
+#include <map>
+#include <optional>
+#include <set>
+#include <thread>
+#include <vector>
+
+#include "io/address.hpp"
+#include "io/rsm/raft.hpp"
+#include "io/rsm/rsm_client.hpp"
+#include "io/simulator/simulator.hpp"
+#include "io/simulator/simulator_transport.hpp"
+
+using memgraph::io::Address;
+using memgraph::io::Duration;
+using memgraph::io::Io;
+using memgraph::io::ResponseEnvelope;
+using memgraph::io::ResponseFuture;
+using memgraph::io::ResponseResult;
+using memgraph::io::Time;
+using memgraph::io::rsm::Raft;
+using memgraph::io::rsm::ReadRequest;
+using memgraph::io::rsm::ReadResponse;
+using memgraph::io::rsm::RsmClient;
+using memgraph::io::rsm::WriteRequest;
+using memgraph::io::rsm::WriteResponse;
+using memgraph::io::simulator::Simulator;
+using memgraph::io::simulator::SimulatorConfig;
+using memgraph::io::simulator::SimulatorStats;
+using memgraph::io::simulator::SimulatorTransport;
+
+struct CasRequest {
+  int key;
+  std::optional<int> old_value;
+  std::optional<int> new_value;
+};
+
+struct CasResponse {
+  bool cas_success;
+  std::optional<int> last_value;
+};
+
+struct GetRequest {
+  int key;
+};
+
+struct GetResponse {
+  std::optional<int> value;
+};
+
+class TestState {
+  std::map<int, int> state_;
+
+ public:
+  GetResponse Read(GetRequest request) {
+    GetResponse ret;
+    if (state_.contains(request.key)) {
+      ret.value = state_[request.key];
+    }
+    return ret;
+  }
+
+  CasResponse Apply(CasRequest request) {
+    CasResponse ret;
+
+    // Key exist
+    if (state_.contains(request.key)) {
+      auto &val = state_[request.key];
+
+      /*
+       *   Delete
+       */
+      if (!request.new_value) {
+        ret.last_value = val;
+        ret.cas_success = true;
+
+        state_.erase(state_.find(request.key));
+      }
+
+      /*
+       *   Update
+       */
+      // Does old_value match?
+      if (request.old_value == val) {
+        ret.last_value = val;
+        ret.cas_success = true;
+
+        val = request.new_value.value();
+      } else {
+        ret.last_value = val;
+        ret.cas_success = false;
+      }
+    }
+    /*
+     *   Create
+     */
+    else {
+      ret.last_value = std::nullopt;
+      ret.cas_success = true;
+
+      state_.emplace(request.key, std::move(request.new_value).value());
+    }
+
+    return ret;
+  }
+};
+
+template <typename IoImpl>
+void RunRaft(Raft<IoImpl, TestState, CasRequest, CasResponse, GetRequest, GetResponse> server) {
+  server.Run();
+}
+
+void RunSimulation() {
+  SimulatorConfig config{
+      .drop_percent = 5,
+      .perform_timeouts = true,
+      .scramble_messages = true,
+      .rng_seed = 0,
+      .start_time = Time::min() + std::chrono::microseconds{256 * 1024},
+      .abort_time = Time::min() + std::chrono::microseconds{8 * 1024 * 1024},
+  };
+
+  auto simulator = Simulator(config);
+
+  auto cli_addr = Address::TestAddress(1);
+  auto srv_addr_1 = Address::TestAddress(2);
+  auto srv_addr_2 = Address::TestAddress(3);
+  auto srv_addr_3 = Address::TestAddress(4);
+
+  Io<SimulatorTransport> cli_io = simulator.Register(cli_addr);
+  Io<SimulatorTransport> srv_io_1 = simulator.Register(srv_addr_1);
+  Io<SimulatorTransport> srv_io_2 = simulator.Register(srv_addr_2);
+  Io<SimulatorTransport> srv_io_3 = simulator.Register(srv_addr_3);
+
+  std::vector<Address> srv_1_peers = {srv_addr_2, srv_addr_3};
+  std::vector<Address> srv_2_peers = {srv_addr_1, srv_addr_3};
+  std::vector<Address> srv_3_peers = {srv_addr_1, srv_addr_2};
+
+  using RaftClass = Raft<SimulatorTransport, TestState, CasRequest, CasResponse, GetRequest, GetResponse>;
+  RaftClass srv_1{std::move(srv_io_1), srv_1_peers, TestState{}};
+  RaftClass srv_2{std::move(srv_io_2), srv_2_peers, TestState{}};
+  RaftClass srv_3{std::move(srv_io_3), srv_3_peers, TestState{}};
+
+  auto srv_thread_1 = std::jthread(RunRaft<SimulatorTransport>, std::move(srv_1));
+  simulator.IncrementServerCountAndWaitForQuiescentState(srv_addr_1);
+
+  auto srv_thread_2 = std::jthread(RunRaft<SimulatorTransport>, std::move(srv_2));
+  simulator.IncrementServerCountAndWaitForQuiescentState(srv_addr_2);
+
+  auto srv_thread_3 = std::jthread(RunRaft<SimulatorTransport>, std::move(srv_3));
+  simulator.IncrementServerCountAndWaitForQuiescentState(srv_addr_3);
+
+  spdlog::debug("beginning test after servers have become quiescent");
+
+  std::mt19937 cli_rng_{0};
+  std::vector<Address> server_addrs{srv_addr_1, srv_addr_2, srv_addr_3};
+  Address leader = server_addrs[0];
+
+  RsmClient<Io<SimulatorTransport>, CasRequest, CasResponse, GetRequest, GetResponse> client(cli_io, leader,
+                                                                                             server_addrs);
+
+  const int key = 0;
+  std::optional<int> last_known_value;
+
+  bool success = false;
+
+  for (int i = 0; !success; i++) {
+    /*
+     * Write Request
+     */
+    CasRequest cas_req;
+    cas_req.key = key;
+
+    cas_req.old_value = last_known_value;
+
+    cas_req.new_value = i;
+
+    auto write_cas_response_result = client.SendWriteRequest(cas_req);
+    if (write_cas_response_result.HasError()) {
+      // timed out
+      continue;
+    }
+    CasResponse cas_response = write_cas_response_result.GetValue();
+
+    bool cas_succeeded = cas_response.cas_success;
+
+    spdlog::debug("Client received CasResponse! success: {} last_known_value {}", cas_succeeded,
+                  (int)*last_known_value);
+
+    if (cas_succeeded) {
+      last_known_value = i;
+    } else {
+      last_known_value = cas_response.last_value;
+      continue;
+    }
+
+    /*
+     * Get Request
+     */
+    GetRequest get_req;
+    get_req.key = key;
+
+    auto read_get_response_result = client.SendReadRequest(get_req);
+    if (read_get_response_result.HasError()) {
+      // timed out
+      continue;
+    }
+    GetResponse get_response = read_get_response_result.GetValue();
+
+    MG_ASSERT(get_response.value == i);
+
+    spdlog::debug("client successfully cas'd a value and read it back! value: {}", i);
+
+    success = true;
+  }
+
+  MG_ASSERT(success);
+
+  simulator.ShutDown();
+
+  SimulatorStats stats = simulator.Stats();
+
+  spdlog::debug("total messages:     ", stats.total_messages);
+  spdlog::debug("dropped messages:   ", stats.dropped_messages);
+  spdlog::debug("timed out requests: ", stats.timed_out_requests);
+  spdlog::debug("total requests:     ", stats.total_requests);
+  spdlog::debug("total responses:    ", stats.total_responses);
+  spdlog::debug("simulator ticks:    ", stats.simulator_ticks);
+
+  spdlog::debug("========================== SUCCESS :) ==========================");
+
+  /*
+  this is implicit in jthread's dtor
+  srv_thread_1.join();
+  srv_thread_2.join();
+  srv_thread_3.join();
+  */
+}
+
+int main() {
+  int n_tests = 50;
+
+  for (int i = 0; i < n_tests; i++) {
+    spdlog::debug("========================== NEW SIMULATION {} ==========================", i);
+    spdlog::debug("\tTime\tTerm\tPort\tRole\t\tMessage\n");
+    RunSimulation();
+  }
+
+  spdlog::debug("passed {} tests!", n_tests);
+
+  return 0;
+}

tests/simulation/sharded_map.cpp

@@ -0,0 +1,337 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#include <chrono>
+#include <deque>
+#include <iostream>
+#include <map>
+#include <optional>
+#include <set>
+#include <thread>
+#include <vector>
+
+#include "io/address.hpp"
+#include "io/errors.hpp"
+#include "io/rsm/coordinator_rsm.hpp"
+#include "io/rsm/raft.hpp"
+#include "io/rsm/rsm_client.hpp"
+#include "io/rsm/shard_rsm.hpp"
+#include "io/simulator/simulator.hpp"
+#include "io/simulator/simulator_transport.hpp"
+#include "utils/result.hpp"
+
+using memgraph::coordinator::Address;
+using memgraph::coordinator::AddressAndStatus;
+using memgraph::coordinator::CompoundKey;
+using memgraph::coordinator::Coordinator;
+using memgraph::coordinator::HlcRequest;
+using memgraph::coordinator::HlcResponse;
+using memgraph::coordinator::Shard;
+using memgraph::coordinator::ShardMap;
+using memgraph::coordinator::Shards;
+using memgraph::coordinator::Status;
+using memgraph::io::Address;
+using memgraph::io::Io;
+using memgraph::io::ResponseEnvelope;
+using memgraph::io::ResponseFuture;
+using memgraph::io::Time;
+using memgraph::io::TimedOut;
+using memgraph::io::rsm::CoordinatorRsm;
+using memgraph::io::rsm::Raft;
+using memgraph::io::rsm::ReadRequest;
+using memgraph::io::rsm::ReadResponse;
+using memgraph::io::rsm::RsmClient;
+using memgraph::io::rsm::StorageReadRequest;
+using memgraph::io::rsm::StorageReadResponse;
+using memgraph::io::rsm::StorageRsm;
+using memgraph::io::rsm::StorageWriteRequest;
+using memgraph::io::rsm::StorageWriteResponse;
+using memgraph::io::rsm::WriteRequest;
+using memgraph::io::rsm::WriteResponse;
+using memgraph::io::simulator::Simulator;
+using memgraph::io::simulator::SimulatorConfig;
+using memgraph::io::simulator::SimulatorStats;
+using memgraph::io::simulator::SimulatorTransport;
+using memgraph::utils::BasicResult;
+
+using StorageClient = RsmClient<Io<SimulatorTransport>, StorageWriteRequest, StorageWriteResponse, StorageReadRequest,
+                                StorageReadResponse>;
+namespace {
+ShardMap CreateDummyShardmap(memgraph::coordinator::Address a_io_1, memgraph::coordinator::Address a_io_2,
+                             memgraph::coordinator::Address a_io_3, memgraph::coordinator::Address b_io_1,
+                             memgraph::coordinator::Address b_io_2, memgraph::coordinator::Address b_io_3) {
+  ShardMap sm1;
+  auto &shards = sm1.GetShards();
+
+  // 1
+  std::string label1 = std::string("label1");
+  auto key1 = memgraph::storage::v3::PropertyValue(3);
+  auto key2 = memgraph::storage::v3::PropertyValue(4);
+  CompoundKey cm1 = {key1, key2};
+  AddressAndStatus aas1_1{.address = a_io_1, .status = Status::CONSENSUS_PARTICIPANT};
+  AddressAndStatus aas1_2{.address = a_io_2, .status = Status::CONSENSUS_PARTICIPANT};
+  AddressAndStatus aas1_3{.address = a_io_3, .status = Status::CONSENSUS_PARTICIPANT};
+
+  Shard shard1 = {aas1_1, aas1_2, aas1_3};
+  Shards shards1;
+  shards1[cm1] = shard1;
+
+  // 2
+  std::string label2 = std::string("label2");
+  auto key3 = memgraph::storage::v3::PropertyValue(12);
+  auto key4 = memgraph::storage::v3::PropertyValue(13);
+  CompoundKey cm2 = {key3, key4};
+  AddressAndStatus aas2_1{.address = b_io_1, .status = Status::CONSENSUS_PARTICIPANT};
+  AddressAndStatus aas2_2{.address = b_io_2, .status = Status::CONSENSUS_PARTICIPANT};
+  AddressAndStatus aas2_3{.address = b_io_3, .status = Status::CONSENSUS_PARTICIPANT};
+
+  Shard shard2 = {aas2_1, aas2_2, aas2_3};
+  Shards shards2;
+  shards2[cm2] = shard2;
+
+  shards[label1] = shards1;
+  shards[label2] = shards2;
+
+  return sm1;
+}
+
+std::optional<StorageClient> DetermineShardLocation(Shard target_shard, const std::vector<Address> &a_addrs,
+                                                    StorageClient a_client, const std::vector<Address> &b_addrs,
+                                                    StorageClient b_client) {
+  for (const auto &addr : target_shard) {
+    if (addr.address == b_addrs[0]) {
+      return b_client;
+    }
+    if (addr.address == a_addrs[0]) {
+      return a_client;
+    }
+  }
+  return {};
+}
+
+}  // namespace
+
+using ConcreteCoordinatorRsm = CoordinatorRsm<SimulatorTransport>;
+using ConcreteStorageRsm = Raft<SimulatorTransport, StorageRsm, StorageWriteRequest, StorageWriteResponse,
+                                StorageReadRequest, StorageReadResponse>;
+
+template <typename IoImpl>
+void RunStorageRaft(
+    Raft<IoImpl, StorageRsm, StorageWriteRequest, StorageWriteResponse, StorageReadRequest, StorageReadResponse>
+        server) {
+  server.Run();
+}
+
+int main() {
+  SimulatorConfig config{
+      .drop_percent = 5,
+      .perform_timeouts = true,
+      .scramble_messages = true,
+      .rng_seed = 0,
+      .start_time = Time::min() + std::chrono::microseconds{256 * 1024},
+      .abort_time = Time::min() + std::chrono::microseconds{2 * 8 * 1024 * 1024},
+  };
+
+  auto simulator = Simulator(config);
+
+  Io<SimulatorTransport> cli_io = simulator.RegisterNew();
+
+  // Register
+  Io<SimulatorTransport> a_io_1 = simulator.RegisterNew();
+  Io<SimulatorTransport> a_io_2 = simulator.RegisterNew();
+  Io<SimulatorTransport> a_io_3 = simulator.RegisterNew();
+
+  Io<SimulatorTransport> b_io_1 = simulator.RegisterNew();
+  Io<SimulatorTransport> b_io_2 = simulator.RegisterNew();
+  Io<SimulatorTransport> b_io_3 = simulator.RegisterNew();
+
+  // Preconfigure coordinator with kv shard 'A' and 'B'
+  auto sm1 = CreateDummyShardmap(a_io_1.GetAddress(), a_io_2.GetAddress(), a_io_3.GetAddress(), b_io_1.GetAddress(),
+                                 b_io_2.GetAddress(), b_io_3.GetAddress());
+  auto sm2 = CreateDummyShardmap(a_io_1.GetAddress(), a_io_2.GetAddress(), a_io_3.GetAddress(), b_io_1.GetAddress(),
+                                 b_io_2.GetAddress(), b_io_3.GetAddress());
+  auto sm3 = CreateDummyShardmap(a_io_1.GetAddress(), a_io_2.GetAddress(), a_io_3.GetAddress(), b_io_1.GetAddress(),
+                                 b_io_2.GetAddress(), b_io_3.GetAddress());
+
+  // Spin up shard A
+  std::vector<Address> a_addrs = {a_io_1.GetAddress(), a_io_2.GetAddress(), a_io_3.GetAddress()};
+
+  std::vector<Address> a_1_peers = {a_addrs[1], a_addrs[2]};
+  std::vector<Address> a_2_peers = {a_addrs[0], a_addrs[2]};
+  std::vector<Address> a_3_peers = {a_addrs[0], a_addrs[1]};
+
+  ConcreteStorageRsm a_1{std::move(a_io_1), a_1_peers, StorageRsm{}};
+  ConcreteStorageRsm a_2{std::move(a_io_2), a_2_peers, StorageRsm{}};
+  ConcreteStorageRsm a_3{std::move(a_io_3), a_3_peers, StorageRsm{}};
+
+  auto a_thread_1 = std::jthread(RunStorageRaft<SimulatorTransport>, std::move(a_1));
+  simulator.IncrementServerCountAndWaitForQuiescentState(a_addrs[0]);
+
+  auto a_thread_2 = std::jthread(RunStorageRaft<SimulatorTransport>, std::move(a_2));
+  simulator.IncrementServerCountAndWaitForQuiescentState(a_addrs[1]);
+
+  auto a_thread_3 = std::jthread(RunStorageRaft<SimulatorTransport>, std::move(a_3));
+  simulator.IncrementServerCountAndWaitForQuiescentState(a_addrs[2]);
+
+  // Spin up shard B
+  std::vector<Address> b_addrs = {b_io_1.GetAddress(), b_io_2.GetAddress(), b_io_3.GetAddress()};
+
+  std::vector<Address> b_1_peers = {b_addrs[1], b_addrs[2]};
+  std::vector<Address> b_2_peers = {b_addrs[0], b_addrs[2]};
+  std::vector<Address> b_3_peers = {b_addrs[0], b_addrs[1]};
+
+  ConcreteStorageRsm b_1{std::move(b_io_1), b_1_peers, StorageRsm{}};
+  ConcreteStorageRsm b_2{std::move(b_io_2), b_2_peers, StorageRsm{}};
+  ConcreteStorageRsm b_3{std::move(b_io_3), b_3_peers, StorageRsm{}};
+
+  auto b_thread_1 = std::jthread(RunStorageRaft<SimulatorTransport>, std::move(b_1));
+  simulator.IncrementServerCountAndWaitForQuiescentState(b_addrs[0]);
+
+  auto b_thread_2 = std::jthread(RunStorageRaft<SimulatorTransport>, std::move(b_2));
+  simulator.IncrementServerCountAndWaitForQuiescentState(b_addrs[1]);
+
+  auto b_thread_3 = std::jthread(RunStorageRaft<SimulatorTransport>, std::move(b_3));
+  simulator.IncrementServerCountAndWaitForQuiescentState(b_addrs[2]);
+
+  // Spin up coordinators
+
+  Io<SimulatorTransport> c_io_1 = simulator.RegisterNew();
+  Io<SimulatorTransport> c_io_2 = simulator.RegisterNew();
+  Io<SimulatorTransport> c_io_3 = simulator.RegisterNew();
+
+  std::vector<Address> c_addrs = {c_io_1.GetAddress(), c_io_2.GetAddress(), c_io_3.GetAddress()};
+
+  std::vector<Address> c_1_peers = {c_addrs[1], c_addrs[2]};
+  std::vector<Address> c_2_peers = {c_addrs[0], c_addrs[2]};
+  std::vector<Address> c_3_peers = {c_addrs[0], c_addrs[1]};
+
+  ConcreteCoordinatorRsm c_1{std::move(c_io_1), c_1_peers, Coordinator{(sm1)}};
+  ConcreteCoordinatorRsm c_2{std::move(c_io_2), c_2_peers, Coordinator{(sm2)}};
+  ConcreteCoordinatorRsm c_3{std::move(c_io_3), c_3_peers, Coordinator{(sm3)}};
+
+  auto c_thread_1 = std::jthread([c_1]() mutable { c_1.Run(); });
+  simulator.IncrementServerCountAndWaitForQuiescentState(c_addrs[0]);
+
+  auto c_thread_2 = std::jthread([c_2]() mutable { c_2.Run(); });
+  simulator.IncrementServerCountAndWaitForQuiescentState(c_addrs[1]);
+
+  auto c_thread_3 = std::jthread([c_3]() mutable { c_3.Run(); });
+  simulator.IncrementServerCountAndWaitForQuiescentState(c_addrs[2]);
+
+  // Have client contact coordinator RSM for a new transaction ID and
+  // also get the current shard map
+  using CoordinatorClient =
+      RsmClient<Io<SimulatorTransport>, memgraph::coordinator::WriteRequests, memgraph::coordinator::WriteResponses,
+                memgraph::coordinator::ReadRequests, memgraph::coordinator::ReadResponses>;
+  CoordinatorClient coordinator_client(cli_io, c_addrs[0], c_addrs);
+
+  StorageClient shard_a_client(cli_io, a_addrs[0], a_addrs);
+  StorageClient shard_b_client(cli_io, b_addrs[0], b_addrs);
+
+  memgraph::coordinator::HlcRequest req;
+
+  // Last ShardMap Version The query engine knows about.
+  ShardMap client_shard_map;
+  req.last_shard_map_version = client_shard_map.GetHlc();
+
+  while (true) {
+    // Create CompoundKey
+    const auto cm_key_1 = memgraph::storage::v3::PropertyValue(3);
+    const auto cm_key_2 = memgraph::storage::v3::PropertyValue(4);
+
+    const CompoundKey cm_k = {cm_key_1, cm_key_2};
+
+    // Look for Shard
+    BasicResult<TimedOut, memgraph::coordinator::ReadResponses> read_res = coordinator_client.SendReadRequest(req);
+
+    if (read_res.HasError()) {
+      // timeout
+      continue;
+    }
+
+    auto coordinator_read_response = read_res.GetValue();
+    HlcResponse hlc_response = std::get<HlcResponse>(coordinator_read_response);
+
+    // Transaction ID to be used later...
+    auto transaction_id = hlc_response.new_hlc;
+
+    if (hlc_response.fresher_shard_map) {
+      client_shard_map = hlc_response.fresher_shard_map.value();
+    }
+
+    // TODO(gabor) check somewhere in the call chain if the entries are actually valid
+    // for (auto &[key, val] : client_shard_map.GetShards()) {
+    //   std::cout << "key: " << key << std::endl;
+    // }
+
+    auto target_shard = client_shard_map.GetShardForKey(std::string("label1"), cm_k);
+
+    // Determine which shard to send the requests to
+    auto storage_client_opt = DetermineShardLocation(target_shard, a_addrs, shard_a_client, b_addrs, shard_b_client);
+    MG_ASSERT(storage_client_opt);
+
+    auto storage_client = storage_client_opt.value();
+
+    // Have client use shard map to decide which shard to communicate
+    // with in order to write a new value
+    // client_shard_map.
+    auto write_key_1 = memgraph::storage::PropertyValue(3);
+    auto write_key_2 = memgraph::storage::PropertyValue(4);
+
+    StorageWriteRequest storage_req;
+    storage_req.key = {write_key_1, write_key_2};
+    storage_req.value = 1000;
+
+    auto write_response_result = storage_client.SendWriteRequest(storage_req);
+    if (write_response_result.HasError()) {
+      // timed out
+      continue;
+    }
+    auto write_response = write_response_result.GetValue();
+
+    bool cas_succeeded = write_response.shard_rsm_success;
+
+    if (!cas_succeeded) {
+      continue;
+    }
+    // Have client use shard map to decide which shard to communicate
+    // with to read that same value back
+
+    StorageReadRequest storage_get_req;
+    storage_get_req.key = {write_key_1, write_key_2};
+
+    auto get_response_result = storage_client.SendReadRequest(storage_get_req);
+    if (get_response_result.HasError()) {
+      // timed out
+      continue;
+    }
+    auto get_response = get_response_result.GetValue();
+    auto val = get_response.value.value();
+
+    MG_ASSERT(val == 1000);
+    break;
+  }
+
+  simulator.ShutDown();
+
+  SimulatorStats stats = simulator.Stats();
+
+  std::cout << "total messages:     " << stats.total_messages << std::endl;
+  std::cout << "dropped messages:   " << stats.dropped_messages << std::endl;
+  std::cout << "timed out requests: " << stats.timed_out_requests << std::endl;
+  std::cout << "total requests:     " << stats.total_requests << std::endl;
+  std::cout << "total responses:    " << stats.total_responses << std::endl;
+  std::cout << "simulator ticks:    " << stats.simulator_ticks << std::endl;
+
+  std::cout << "========================== SUCCESS :) ==========================" << std::endl;
+
+  return 0;
+}

tests/unit/CMakeLists.txt

@@ -1,3 +1,5 @@
+include(MgThrift)
+
 set(test_prefix memgraph__unit__)
 
 find_package(fmt REQUIRED)
@@ -391,12 +393,14 @@ add_custom_target(test_lcp ALL DEPENDS ${CMAKE_CURRENT_BINARY_DIR}/test_lcp)
 add_test(test_lcp ${CMAKE_CURRENT_BINARY_DIR}/test_lcp)
 add_dependencies(memgraph__unit test_lcp)
 
-# Test websocket
-find_package(Boost REQUIRED)
-
-add_unit_test(websocket.cpp)
-target_link_libraries(${test_prefix}websocket mg-communication Boost::headers)
-
 # Test future
 add_unit_test(future.cpp)
-target_link_libraries(${test_prefix}future mg-io)
+target_link_libraries(${test_prefix}future mg-io)
+
+# Test Thrift transport echo
+add_unit_test(thrift_transport_echo.cpp)
+target_link_libraries(${test_prefix}thrift_transport_echo mg-io fmt Threads::Threads FBThrift::thriftcpp2 mg-interface-echo-cpp2)
+
+# Test Thrift transport echo
+add_unit_test(thrift_handle.cpp)
+target_link_libraries(${test_prefix}thrift_handle mg-io fmt Threads::Threads)

tests/unit/cla_helper.hpp

@@ -0,0 +1,24 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#include <string>
+#include "gtest/gtest.h"
+
+namespace clahelper {
+inline std::string g_command_line_arg;
+}
+
+class InjectedClaEnvironment : public testing::Environment {
+ public:
+  explicit InjectedClaEnvironment(const std::string &command_line_arg) {
+    clahelper::g_command_line_arg = command_line_arg;
+  }
+};

tests/unit/main.cpp

@@ -12,8 +12,15 @@
 #include <gtest/gtest.h>
 #include <utils/logging.hpp>
 
+#include "cla_helper.hpp"
+
 int main(int argc, char **argv) {
   ::testing::InitGoogleTest(&argc, argv);
+
+  std::string command_line_arg(argv[0]);
+  testing::InitGoogleTest(&argc, argv);
+  testing::AddGlobalTestEnvironment(new InjectedClaEnvironment(command_line_arg));
+
   memgraph::logging::RedirectToStderr();
   spdlog::set_level(spdlog::level::trace);
   return RUN_ALL_TESTS();

tests/unit/thrift_handle.cpp

@@ -0,0 +1,108 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#include <chrono>
+#include <string>
+#include <thread>
+
+#include "gtest/gtest.h"
+
+#include "io/address.hpp"
+#include "io/future.hpp"
+#include "io/thrift/thrift_handle.hpp"
+#include "io/transport.hpp"
+#include "utils/logging.hpp"
+
+using memgraph::io::Address;
+using memgraph::io::Duration;
+using memgraph::io::FuturePromisePair;
+using memgraph::io::RequestEnvelope;
+using memgraph::io::ResponseEnvelope;
+using memgraph::io::ResponseResult;
+using memgraph::io::Time;
+using memgraph::io::thrift::ThriftHandle;
+
+namespace {
+struct TestMessage {
+  int value;
+};
+}  // namespace
+
+TEST(Thrift, ThriftHandleTimeout) {
+  auto our_address = Address::TestAddress(0);
+  auto handle = ThriftHandle{our_address};
+
+  // assert timeouts fire
+  Duration zero_timeout = Duration{};
+  auto should_timeout_1 = handle.Receive<TestMessage>(zero_timeout);
+  MG_ASSERT(should_timeout_1.HasError());
+
+  Duration ten_ms = std::chrono::microseconds{10000};
+  Time before = handle.Now();
+
+  auto should_timeout_2 = handle.Receive<TestMessage>(ten_ms);
+  MG_ASSERT(should_timeout_2.HasError());
+
+  Time after = handle.Now();
+
+  MG_ASSERT(after - before >= ten_ms);
+}
+
+TEST(Thrift, ThriftHandleReceive) {
+  auto our_address = Address::TestAddress(0);
+  auto handle = ThriftHandle{our_address};
+
+  // assert we can send and receive
+  auto to_address = Address::TestAddress(0);
+  auto from_address = Address::TestAddress(1);
+  auto request_id = 0;
+  auto message = TestMessage{
+      .value = 777,
+  };
+
+  handle.DeliverMessage(to_address, from_address, request_id, std::move(message));
+
+  auto should_have_message = handle.Receive<TestMessage>(Duration{});
+  MG_ASSERT(should_have_message.HasValue());
+
+  RequestEnvelope<TestMessage> re = should_have_message.GetValue();
+  TestMessage request = std::get<TestMessage>(std::move(re.message));
+  MG_ASSERT(request.value == 777);
+}
+
+/// this test "sends" a TestMessage to a server and expects to receive
+/// a TestMessage back with the same value.
+TEST(Thrift, ThriftHandleRequestReceive) {
+  // use the same address for now, to rely on loopback optimization
+  auto our_address = Address::TestAddress(0);
+  auto cli_address = our_address;
+  auto srv_address = cli_address;
+
+  auto handle = ThriftHandle{our_address};
+
+  auto timeout = Duration{};
+  auto request_id = 1;
+  auto expected_value = 323;
+  auto request = TestMessage{};
+  request.value = expected_value;
+
+  auto [future, promise] = FuturePromisePair<ResponseResult<TestMessage>>();
+
+  handle.SubmitRequest(srv_address, cli_address, request_id, std::move(request), timeout, std::move(promise));
+
+  // TODO(tyler) do actual socket stuff in the future maybe
+
+  ResponseResult<TestMessage> response_result = std::move(future).Wait();
+  MG_ASSERT(response_result.HasValue());
+  ResponseEnvelope<TestMessage> response_envelope = response_result.GetValue();
+  TestMessage response = response_envelope.message;
+  MG_ASSERT(response.value == expected_value);
+}

tests/unit/thrift_transport_echo.cpp

@@ -0,0 +1,132 @@
+// Copyright 2022 Memgraph Ltd.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt; by using this file, you agree to be bound by the terms of the Business Source
+// License, and you may not use this file except in compliance with the Business Source License.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+#include <string>
+#include <thread>
+
+#include "gtest/gtest.h"
+
+#include <folly/init/Init.h>
+#include <folly/io/SocketOptionMap.h>
+#include <folly/io/async/AsyncServerSocket.h>
+#include <folly/net/NetworkSocket.h>
+#include <thrift/lib/cpp2/async/HeaderClientChannel.h>
+#include <thrift/lib/cpp2/server/ThriftServer.h>
+#include "interface/gen-cpp2/Echo.h"  // From generated code
+#include "interface/gen-cpp2/EchoAsyncClient.h"
+
+#include "io/thrift/thrift_transport.hpp"
+
+#include "cla_helper.hpp"
+
+using namespace memgraph::io;
+
+using namespace apache::thrift;
+using namespace cpp2;
+using namespace folly;
+
+using namespace std::chrono_literals;
+
+// static constexpr int port = 6666;  // The port on which server is listening
+
+class EchoSvc : public EchoSvIf {
+  inline static const std::string prefix_{"0"};
+
+  std::string current_message_;
+  // bool has_message_{false};
+
+ public:
+  virtual ~EchoSvc() {}
+
+  // The Thrift handle method
+  void ReceiveSend(const EchoMessage &m) override {
+    // m.get_message();
+    // LOG(ERROR) << "Received\n";
+    current_message_ = prefix_ + m.get_message();
+    // SendOutMessage(6665);
+    //  LOG(ERROR) << "Sent\n";
+  }
+
+  void SendOneShotMessage(int other_port, const std::string &message_str) {
+    EventBase base;
+    auto socket(folly::AsyncSocket::newSocket(&base, "127.0.0.1", other_port));
+
+    // Create a HeaderClientChannel object which is used in creating
+    // client object
+    auto client_channel = HeaderClientChannel::newChannel(std::move(socket));
+    // Create a client object
+    EchoAsyncClient client(std::move(client_channel));
+
+    EchoMessage message;
+    message.message_ref() = message_str;
+    client.sync_ReceiveSend(message);
+  }
+
+  void SendOutMessage(int other_port) {
+    EventBase base;
+    auto socket(folly::AsyncSocket::newSocket(&base, "127.0.0.1", other_port));
+
+    // Create a HeaderClientChannel object which is used in creating
+    // client object
+    auto client_channel = HeaderClientChannel::newChannel(std::move(socket));
+    // Create a client object
+    EchoAsyncClient client(std::move(client_channel));
+
+    EchoMessage message;
+    message.message_ref() = current_message_;
+    client.sync_ReceiveSend(message);
+  }
+
+  std::string GetCurrentMessage() { return current_message_; }
+};
+
+TEST(ThriftTransport, Echo) {
+  // TODO(tyler and gabor) use thrift-generated echo, and thrift transport, to send, reply, and receive the response for
+  // a thrift-defined message
+  int argc = 1;
+  const char *arg2 = clahelper::g_command_line_arg.c_str();
+  char *arg = const_cast<char *>(arg2);
+  char **argv = &arg;
+  folly::Init(&argc, &argv);
+
+  auto ptr1 = std::make_shared<EchoSvc>();
+  auto ptr2 = std::make_shared<EchoSvc>();
+
+  auto server_thread2 = std::jthread([&ptr2] {
+    ThriftServer *s = new ThriftServer();
+    s->setInterface(ptr2);
+    s->setPort(6666);
+    s->serve();
+  });
+
+  auto server_thread1 = std::jthread([&ptr1] {
+    ThriftServer *s = new ThriftServer();
+    s->setInterface(ptr1);
+    s->setPort(6665);
+    s->serve();
+  });
+
+  std::this_thread::sleep_for(4000ms);
+
+  ptr1->SendOneShotMessage(6666, "original");
+  std::this_thread::sleep_for(4000ms);
+
+  ptr2->SendOutMessage(6665);
+  std::this_thread::sleep_for(4000ms);
+
+  auto result = ptr1->GetCurrentMessage();
+
+  ASSERT_EQ(result, std::string("00original"));
+
+  // Solve this once this is not just a POC.
+  server_thread2.detach();
+  server_thread1.detach();
+}