statetrack_plugin implementation

.gitmodules

@@ -1,6 +1,6 @@
 [submodule "libraries/chainbase"]
 	path = libraries/chainbase
-	url = https://github.com/eosio/chainbase
+	url = https://github.com/mmcs85/chainbase
         ignore = dirty
 [submodule "libraries/appbase"]
 	path = libraries/appbase

plugins/CMakeLists.txt

@@ -7,6 +7,7 @@ add_subdirectory(chain_plugin)
 add_subdirectory(chain_api_plugin)
 add_subdirectory(producer_plugin)
 add_subdirectory(producer_api_plugin)
+add_subdirectory(statetrack_plugin)
 add_subdirectory(history_plugin)
 add_subdirectory(history_api_plugin)
 

plugins/statetrack_plugin/CMakeLists.txt

@@ -0,0 +1,29 @@
+list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_SOURCE_DIR}/CMakeModules")
+
+## load in pkg-config support
+find_package(PkgConfig REQUIRED)
+## use pkg-config to get hints for 0mq locations
+pkg_check_modules(PC_ZeroMQ REQUIRED libzmq)
+
+## use the hint from above to find where 'zmq.hpp' is located
+find_path(ZeroMQ_INCLUDE_DIR
+  NAMES zmq.hpp
+  PATHS ${PC_ZeroMQ_INCLUDE_DIRS}
+)
+
+## use the hint from about to find the location of libzmq
+find_library(ZeroMQ_LIBRARY
+  NAMES zmq
+  PATHS ${PC_ZeroMQ_LIBRARY_DIRS}
+  )
+
+message(STATUS "[Additional Plugin] EOSIO ZeroMQ plugin enabled")
+
+file(GLOB HEADERS "include/eosio/statetrack_plugin/*.hpp")
+add_library( statetrack_plugin
+             statetrack_plugin.cpp
+             statetrack_plugin_impl.cpp
+             ${HEADERS} )
+
+target_link_libraries( statetrack_plugin ${ZeroMQ_LIBRARY} chain_plugin appbase fc )
+target_include_directories( statetrack_plugin PUBLIC "${CMAKE_CURRENT_SOURCE_DIR}/include" )

plugins/statetrack_plugin/README.md

@@ -0,0 +1,116 @@
+# statetrack_plugin
+
+This plugin adds hooks on chainbase operations emplace, modify, remove, and undo,
+and routes the objects and revision numbers to a ZMQ socket.
+This approach has different purposes:
+
+- Allowing developers to get fine-grained access to the state DB operations.
+  - To mirror the state DB into a traditional database system such as MongoDB
+    or Postgres without them having to implement any further logic or duplicate
+    the DB write logic of their smart contracts.
+  - To execute side-effects associated to specific database operations rather than
+    specific contract actions.
+  - To propagate database operations to the dapp frontends through WebSockets
+    for enabling real-time updates to user interfaces.
+- Seamlessly handling any forks and transaction failures by leveraging nodeos' own
+  behavior in maintaining the consistency of its own databases through
+  undo and commit operations.
+- Minimizing the work that happens inside nodeos and letting receivers such as
+  [statemirror](https://github.com/andresberrios/statemirror) or a demux-js reader
+  handle the more complex logic and more performance intensive work.
+
+The plugin can hook to the operations that happen on the state DBs of all contracts
+deployed to the chain, and it can use filters to limit the operations
+that should be sent to only the ones happening on the contracts, scopes, and tables
+that the user is interested in tracking, thus limiting the performance impact to a minimum.
+
+The plugin will also send operations happening on accounts and account permissions.
+It will also send a stream of applied actions.
+
+This is a very early stage of the implementation, and more testing is required.
+This should not be considered production ready yet.
+
+## Configuration
+
+The following configuration statements in `config.ini` are recognized:
+
+- `plugin = eosio::statetrack_plugin` -- Enables the state track plugin
+- `st-zmq-sender-bind = ENDPOINT` -- Specifies the PUSH socket connection endpoint.
+  Default value: tcp://127.0.0.1:3000.
+- `st-filter-on = code:scope:table` -- Track DB operations which match code:scope:table.
+- `st-filter-out = code:scope:table` -- Do not track DB operations which match code:scope:table.
+
+## Important notes
+
+- Accounts are sent as DB operations with `(code, scope, table)` being
+  `(system, system, accounts)`.
+  - Disable tracking of accounts using `st-filter-out = system:system:accounts`.
+- Account permissions are sent as DB operations with `(code, scope, table)`
+  being `(system, system, permissions)`.
+  - Disable tracking of account permissions using `st-filter-out = system:system:permissions`.
+- Actions are sent as DB operations with `(code, scope, table)` being
+  `(system, system, actions)`.
+  - Disable tracking of actions using `st-filter-out = system:system:actions`.
+- Filters are not tested yet, but the implementation is taken from
+  the history plugin, so it should work fine.
+- We're adding separate filters for actions to make it possible to
+  filter by specific actions of specific contracts.
+
+## Open questions
+
+- In Chainbase, we are emitting the operations inside an undo just before the
+  operation is actually executed and checked for success
+  ([See here](https://github.com/EOSIO/chainbase/compare/master...mmcs85:master#diff-298563b6c76ef92100c2ea27c06cb08bR390)).
+  - We would like advice on the best way (without degrading performance) to get
+    the item after it is given to the `modify` and `emplace` functions using `std::move`.
+  - We don't like this inconsistency since it is emitting the event before being
+    sure that the operation actually took place, but we think it might not
+    be an issue since those operations are undoing previous ones and it
+    would not be an expected scenario for them to fail unless there was a bug
+    or a database inconsistency (maybe caused by bad locking mechanisms when
+    using multiple threads?).
+- Performance
+  - Serialization options
+    - JSON serializing on the plugin
+    - Sending binary and decoding to JSON in the receiver
+      - Would need to fetch ABIs, which would make it slower overall but put less load on nodeos
+    - Place operations in a queue during transaction processing
+      - Process queue, serialize each message to JSON and send to receiver
+        - After transaction is processed (prevents failed transactions from sending do and undo operations)
+        - During transaction processing but in a separate thread
+  - Risk of making transaction processing take longer and hit CPU time limit
+    - Only for transactions sent directly to this node?
+    - Does it affect validated transactions/blocks received from the network?
+- Consensus
+  - Forks
+    - How to induce and test recovery
+  - Exceptions
+  - Other potential disruptions
+- Socket communications
+  - In the current implementation, sending a message blocks until it is received (ZMQ PUSH socket)
+  - This is useful in case receiver goes down, since nodeos will wait for it to come back up
+    - Instead of maintaining a local in-memory queue of messages that can build up too much if it carries on
+    - However, it will make nodeos unavailable for API calls, and maybe other side-effects
+  - Lost messages
+    - When receiver pulls a message from the socket but then crashes
+      - Implement message acknowledgement signal?
+      - Implement an intermediate receiver that only manages the queue?
+      - Implement a very safe persistent received-messages queue in receiver from which to recover if it crashes?
+
+## Building
+
+The plugin needs to be built using the ZeroMQ C++ headers.
+
+### Adding ZMQ library on Mac
+
+```
+brew install zmq
+brew tap jmuncaster/homebrew-header-only
+brew install jmuncaster/header-only/cppzmq
+```
+
+### Adding ZMQ library on Ubuntu
+
+```
+apt-get install -y pkg-config libzmq5-dev
+```

plugins/statetrack_plugin/include/eosio/statetrack_plugin/statetrack_plugin.hpp

@@ -0,0 +1,35 @@
+/**
+ *  @file
+ *  @copyright defined in eos/LICENSE.txt
+ */
+#pragma once
+
+#include <appbase/application.hpp>
+#include <eosio/chain_plugin/chain_plugin.hpp>
+
+namespace eosio {
+
+using namespace appbase;
+
+class statetrack_plugin : public plugin<statetrack_plugin> {
+public:
+   APPBASE_PLUGIN_REQUIRES((chain_plugin))
+
+   statetrack_plugin();
+   statetrack_plugin(const statetrack_plugin&) = delete;
+   statetrack_plugin(statetrack_plugin&&) = delete;
+   statetrack_plugin& operator=(const statetrack_plugin&) = delete;
+   statetrack_plugin& operator=(statetrack_plugin&&) = delete;
+   virtual ~statetrack_plugin() override = default;
+
+   virtual void set_program_options(options_description& cli, options_description& cfg) override;
+   void plugin_initialize(const variables_map& options);
+   void plugin_startup();
+   void plugin_shutdown();
+
+private:
+   std::unique_ptr<class statetrack_plugin_impl> my;
+};
+
+}
+