Add support for block-until-commit responses

CHANGELOG.md

@@ -12,6 +12,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ### Added
 
 - Experimental self-healing-open protocol for automatically transitioning-to-open during a disaster recovery without operator intervention. (#7189)
+- Added support for endpoints which only respond once their content is committed by the consensus protocol (#7562).
 
 ### Changed
 

doc/schemas/app_openapi.json

@@ -257,7 +257,7 @@
   "info": {
     "description": "This CCF sample app implements a simple logging application, securely recording messages at client-specified IDs. It demonstrates most of the features available to CCF apps.",
     "title": "CCF Sample Logging App",
-    "version": "2.8.0"
+    "version": "2.8.1"
   },
   "openapi": "3.0.0",
   "paths": {
@@ -375,6 +375,86 @@
         }
       }
     },
+    "/app/log/blocking/private": {
+      "get": {
+        "operationId": "GetAppLogBlockingPrivate",
+        "parameters": [
+          {
+            "in": "query",
+            "name": "id",
+            "required": true,
+            "schema": {
+              "$ref": "#/components/schemas/uint64"
+            }
+          }
+        ],
+        "responses": {
+          "200": {
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/LoggingGet__Out"
+                }
+              }
+            },
+            "description": "Default response description"
+          },
+          "default": {
+            "$ref": "#/components/responses/default"
+          }
+        },
+        "security": [
+          {
+            "jwt": []
+          },
+          {
+            "user_cose_sign1": []
+          }
+        ],
+        "x-ccf-forwarding": {
+          "$ref": "#/components/x-ccf-forwarding/sometimes"
+        }
+      },
+      "post": {
+        "operationId": "PostAppLogBlockingPrivate",
+        "requestBody": {
+          "content": {
+            "application/json": {
+              "schema": {
+                "$ref": "#/components/schemas/LoggingRecord__In"
+              }
+            }
+          },
+          "description": "Auto-generated request body schema"
+        },
+        "responses": {
+          "200": {
+            "content": {
+              "application/json": {
+                "schema": {
+                  "$ref": "#/components/schemas/boolean"
+                }
+              }
+            },
+            "description": "Default response description"
+          },
+          "default": {
+            "$ref": "#/components/responses/default"
+          }
+        },
+        "security": [
+          {
+            "jwt": []
+          },
+          {
+            "user_cose_sign1": []
+          }
+        ],
+        "x-ccf-forwarding": {
+          "$ref": "#/components/x-ccf-forwarding/always"
+        }
+      }
+    },
     "/app/log/cose_signed_content": {
       "post": {
         "operationId": "PostAppLogCoseSignedContent",

include/ccf/endpoint.h

@@ -259,9 +259,16 @@ namespace ccf::endpoints
   {
     // Functor which is invoked to process requests for this Endpoint
     EndpointFunction func;
-    // Functor which is invoked to modify the response post commit.
+
+    // Functor which is invoked to modify the response after it is locally
+    // committed (ie - assigned a transaction ID)
     LocallyCommittedEndpointFunction locally_committed_func;
 
+    // Functor which is invoked to modify the response after it reaches a
+    // terminal consensus state (ie - it is either globally committed, or
+    // invalidated)
+    ConsensusCommittedEndpointFunction consensus_committed_func;
+
     struct Installer
     {
       virtual void install(Endpoint&) = 0;
@@ -488,6 +495,9 @@ namespace ccf::endpoints
     Endpoint& set_locally_committed_function(
       const LocallyCommittedEndpointFunction& lcf);
 
+    Endpoint& set_consensus_committed_function(
+      const ConsensusCommittedEndpointFunction& ccf_);
+
     void install();
   };
 

include/ccf/endpoint_context.h

@@ -3,6 +3,7 @@
 #pragma once
 
 #include "ccf/endpoints/authentication/authentication_types.h"
+#include "ccf/tx_status.h"
 
 #include <functional>
 #include <memory>
@@ -65,6 +66,11 @@ namespace ccf::endpoints
   using LocallyCommittedEndpointFunction =
     std::function<void(CommandEndpointContext& ctx, const ccf::TxID& txid)>;
 
+  using ConsensusCommittedEndpointFunction = std::function<void(
+    std::shared_ptr<ccf::RpcContext> rpc_ctx,
+    const ccf::TxID& txid,
+    ccf::FinalTxStatus status)>;
+
   // Read-only endpoints can only get values from the kv, they cannot write
   struct ReadOnlyEndpointContext : public CommandEndpointContext
   {

include/ccf/endpoint_registry.h

@@ -60,6 +60,11 @@ namespace ccf::endpoints
   void default_locally_committed_func(
     CommandEndpointContext& ctx, const TxID& tx_id);
 
+  void default_respond_on_commit_func(
+    std::shared_ptr<ccf::RpcContext> rpc_ctx,
+    const TxID& tx_id,
+    ccf::FinalTxStatus status);
+
   template <typename T>
   inline bool get_path_param(
     const ccf::PathParams& params,

include/ccf/tx_status.h

@@ -31,6 +31,13 @@ namespace ccf
     Invalid,
   };
 
+  // Contains only the terminal values of TxStatus
+  enum class FinalTxStatus : uint8_t
+  {
+    Committed = static_cast<uint8_t>(TxStatus::Committed),
+    Invalid = static_cast<uint8_t>(TxStatus::Invalid),
+  };
+
   constexpr char const* tx_status_to_str(TxStatus status)
   {
     switch (status)

samples/apps/logging/logging.cpp

@@ -468,7 +468,7 @@ namespace loggingapp
         "recording messages at client-specified IDs. It demonstrates most of "
         "the features available to CCF apps.";
 
-      openapi_info.document_version = "2.8.0";
+      openapi_info.document_version = "2.8.1";
     };
 
     void init_handlers() override
@@ -517,6 +517,16 @@ namespace loggingapp
         .install();
       // SNIPPET_END: install_record
 
+      make_endpoint(
+        "/log/blocking/private",
+        HTTP_POST,
+        ccf::json_adapter(record),
+        auth_policies)
+        .set_auto_schema<LoggingRecord::In, bool>()
+        .set_consensus_committed_function(
+          ccf::endpoints::default_respond_on_commit_func)
+        .install();
+
       auto add_txid_in_body_put = [](auto& ctx, const auto& tx_id) {
         static constexpr auto CCF_TX_ID = "x-ms-ccf-transaction-id";
         ctx.rpc_ctx->set_response_header(CCF_TX_ID, tx_id.to_str());
@@ -623,6 +633,17 @@ namespace loggingapp
         .install();
       // SNIPPET_END: install_get
 
+      make_read_only_endpoint(
+        "/log/blocking/private",
+        HTTP_GET,
+        ccf::json_read_only_adapter(get),
+        auth_policies)
+        .set_auto_schema<void, LoggingGet::Out>()
+        .add_query_parameter<size_t>("id")
+        .set_consensus_committed_function(
+          ccf::endpoints::default_respond_on_commit_func)
+        .install();
+
       make_read_only_endpoint(
         "/log/private/backup",
         HTTP_GET,

src/consensus/aft/raft.h

@@ -12,6 +12,7 @@
 #include "ds/serialized.h"
 #include "impl/state.h"
 #include "kv/kv_types.h"
+#include "node/commit_callback_subsystem.h"
 #include "node/node_client.h"
 #include "node/node_to_node.h"
 #include "node/node_types.h"
@@ -183,6 +184,8 @@ namespace aft
     // Used to remove retired nodes from store
     std::unique_ptr<ccf::RetiredNodeCleanup> retired_node_cleanup;
 
+    std::shared_ptr<ccf::CommitCallbackSubsystem> commit_callbacks;
+
     size_t entry_size_not_limited = 0;
     size_t entry_count = 0;
     Index entries_batch_size = 20;
@@ -214,6 +217,8 @@ namespace aft
       std::shared_ptr<ccf::NodeToNode> channels_,
       std::shared_ptr<aft::State> state_,
       std::shared_ptr<ccf::NodeClient> rpc_request_context_,
+      std::shared_ptr<ccf::CommitCallbackSubsystem>
+        commit_callbacks_subsystem_ = nullptr,
       bool public_only_ = false) :
       store(std::move(store_)),
 
@@ -228,6 +233,7 @@ namespace aft
       node_client(std::move(rpc_request_context_)),
       retired_node_cleanup(
         std::make_unique<ccf::RetiredNodeCleanup>(node_client)),
+      commit_callbacks(std::move(commit_callbacks_subsystem_)),
 
       public_only(public_only_),
 
@@ -2598,6 +2604,12 @@ namespace aft
       store->compact(idx);
       ledger->commit(idx);
 
+      if (commit_callbacks != nullptr)
+      {
+        const auto term = get_term_internal(idx);
+        commit_callbacks->trigger_callbacks({term, idx});
+      }
+
       RAFT_DEBUG_FMT("Commit on {}: {}", state->node_id, idx);
 
       // Examine each configuration that is followed by a globally committed

src/enclave/enclave.h

@@ -15,6 +15,7 @@
 #include "interface.h"
 #include "js/interpreter_cache.h"
 #include "kv/ledger_chunker.h"
+#include "node/commit_callback_subsystem.h"
 #include "node/historical_queries.h"
 #include "node/network_state.h"
 #include "node/node_state.h"
@@ -143,6 +144,10 @@ namespace ccf
       context->install_subsystem(
         std::make_shared<ccf::AbstractCOSESignaturesConfigSubsystem>(*node));
 
+      auto commit_callbacks = std::make_shared<ccf::CommitCallbackSubsystem>();
+      context->install_subsystem(commit_callbacks);
+      rpcsessions->set_commit_callbacks_subsystem(commit_callbacks);
+
       LOG_TRACE_FMT("Creating RPC actors / ffi");
       rpc_map->register_frontend<ccf::ActorsType::members>(
         std::make_unique<ccf::MemberRpcFrontend>(network, *context));
@@ -160,6 +165,7 @@ namespace ccf
         rpc_map,
         rpcsessions,
         indexer,
+        commit_callbacks,
         sig_tx_interval,
         sig_ms_interval);
     }

src/enclave/rpc_sessions.h

@@ -60,7 +60,10 @@ namespace ccf
     ringbuffer::WriterPtr to_host = nullptr;
     std::shared_ptr<RPCMap> rpc_map;
     std::unordered_map<ListenInterfaceID, std::shared_ptr<::tls::Cert>> certs;
-    std::shared_ptr<CustomProtocolSubsystem> custom_protocol_subsystem;
+    std::shared_ptr<CustomProtocolSubsystem> custom_protocol_subsystem =
+      nullptr;
+    std::shared_ptr<CommitCallbackSubsystem> commit_callbacks_subsystem =
+      nullptr;
 
     ccf::pal::Mutex lock;
     std::unordered_map<
@@ -168,7 +171,8 @@ namespace ccf
           writer_factory,
           std::move(ctx),
           parser_configuration,
-          shared_from_this());
+          shared_from_this(),
+          commit_callbacks_subsystem);
       }
       if (custom_protocol_subsystem)
       {
@@ -186,8 +190,7 @@ namespace ccf
       ringbuffer::AbstractWriterFactory& writer_factory,
       std::shared_ptr<RPCMap> rpc_map_) :
       writer_factory(writer_factory),
-      rpc_map(std::move(rpc_map_)),
-      custom_protocol_subsystem(nullptr)
+      rpc_map(std::move(rpc_map_))
     {
       to_host = writer_factory.create_writer_to_outside();
     }
@@ -198,6 +201,12 @@ namespace ccf
       custom_protocol_subsystem = cpss;
     }
 
+    void set_commit_callbacks_subsystem(
+      std::shared_ptr<CommitCallbackSubsystem> fcss)
+    {
+      commit_callbacks_subsystem = fcss;
+    }
+
     void report_parsing_error(const ccf::ListenInterfaceID& id) override
     {
       std::lock_guard<ccf::pal::Mutex> guard(lock);
@@ -410,7 +419,8 @@ namespace ccf
               writer_factory,
               std::move(ctx),
               per_listen_interface.http_configuration,
-              shared_from_this());
+              shared_from_this(),
+              commit_callbacks_subsystem);
         }
         sessions.insert(std::make_pair(
           id, std::make_pair(listen_interface_id, std::move(capped_session))));

src/endpoints/endpoint.cpp

@@ -103,6 +103,13 @@ namespace ccf::endpoints
     return *this;
   }
 
+  Endpoint& Endpoint::set_consensus_committed_function(
+    const ConsensusCommittedEndpointFunction& ccf_)
+  {
+    consensus_committed_func = ccf_;
+    return *this;
+  }
+
   Endpoint& Endpoint::set_openapi_description(const std::string& description)
   {
     openapi_description = description;

src/endpoints/endpoint_registry.cpp

@@ -201,6 +201,24 @@ namespace ccf::endpoints
     ctx.rpc_ctx->set_response_header(http::headers::CCF_TX_ID, tx_id.to_str());
   }
 
+  void default_respond_on_commit_func(
+    std::shared_ptr<ccf::RpcContext> rpc_ctx,
+    const TxID& tx_id,
+    ccf::FinalTxStatus status)
+  {
+    if (status == ccf::FinalTxStatus::Invalid)
+    {
+      rpc_ctx->set_error(
+        HTTP_STATUS_INTERNAL_SERVER_ERROR,
+        ccf::errors::TransactionInvalid,
+        fmt::format(
+          "While waiting for TxID {} to commit, it was invalidated",
+          tx_id.to_str()));
+    }
+
+    // Else leave the original response untouched, and return it now
+  }
+
   Endpoint EndpointRegistry::make_endpoint(
     const std::string& method,
     RESTVerb verb,