Add support for block-until-commit responses

CHANGELOG.md

@@ -89,6 +89,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,14 @@ namespace ccf
     Invalid,
   };
 
+  // Contains only the terminal values of TxStatus
+  enum class FinalTxStatus : std::underlying_type_t<TxStatus>
+  {
+    Committed =
+      static_cast<std::underlying_type_t<TxStatus>>(TxStatus::Committed),
+    Invalid = static_cast<std::underlying_type_t<TxStatus>>(TxStatus::Invalid),
+  };
+
   constexpr char const* tx_status_to_str(TxStatus status)
   {
     switch (status)

samples/apps/logging/logging.cpp

@@ -512,7 +512,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
@@ -561,6 +561,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());
@@ -667,6 +677,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/impl/state.h

@@ -55,7 +55,7 @@ namespace aft
       LOG_DEBUG_FMT("Resulting views: {}", fmt::join(views, ", "));
     }
 
-    ccf::View view_at(ccf::kv::Version idx)
+    [[nodiscard]] ccf::View view_at(ccf::kv::Version idx) const
     {
       auto it = upper_bound(views.begin(), views.end(), idx);
 
@@ -68,7 +68,7 @@ namespace aft
       return (it - views.begin());
     }
 
-    ccf::kv::Version start_of_view(ccf::View view)
+    [[nodiscard]] ccf::kv::Version start_of_view(ccf::View view) const
     {
       if (view > views.size() || view == InvalidView)
       {
@@ -85,7 +85,7 @@ namespace aft
       return tentative;
     }
 
-    ccf::kv::Version end_of_view(ccf::View view)
+    [[nodiscard]] ccf::kv::Version end_of_view(ccf::View view) const
     {
       // If this view has no start (potentially because it contains no
       // transactions), then it can't have an end
@@ -104,14 +104,15 @@ namespace aft
       return views[view] - 1;
     }
 
-    std::vector<ccf::kv::Version> get_history_until(
-      ccf::kv::Version idx = std::numeric_limits<ccf::kv::Version>::max())
+    [[nodiscard]] std::vector<ccf::kv::Version> get_history_until(
+      ccf::kv::Version idx = std::numeric_limits<ccf::kv::Version>::max()) const
     {
       return {views.begin(), std::upper_bound(views.begin(), views.end(), idx)};
     }
 
     // view should be non-zero as views start at one in here
-    std::vector<ccf::kv::Version> get_history_since(uint64_t view)
+    [[nodiscard]] std::vector<ccf::kv::Version> get_history_since(
+      uint64_t view) const
     {
       if (view > views.size())
       {

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_),
 
@@ -236,7 +242,12 @@ namespace aft
 
       ledger(std::move(ledger_)),
       channels(std::move(channels_))
-    {}
+    {
+      if (commit_callbacks != nullptr)
+      {
+        commit_callbacks->set_consensus(this);
+      }
+    }
 
     ~Aft() override = default;
 
@@ -2602,6 +2613,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}, state->view_history);
+      }
+
       RAFT_DEBUG_FMT("Commit on {}: {}", state->node_id, idx);
 
       // Examine each configuration that is followed by a globally committed

src/enclave/enclave.h

@@ -16,6 +16,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"
@@ -151,6 +152,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));
@@ -168,6 +173,7 @@ namespace ccf
         rpc_map,
         rpcsessions,
         indexer,
+        commit_callbacks,
         sig_tx_interval,
         sig_ms_interval);
     }