action.h

Go to the documentation of this file.

/*
 * Copyright 2026 The A11 Authors
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
 
#ifndef A11_ACTIONS_ACTION_H_
#define A11_ACTIONS_ACTION_H_
 
#include <cstddef>
#include <functional>
#include <memory>
#include <optional>
#include <string>
#include <string_view>
#include <type_traits>
#include <utility>
#include <vector>
 
#include <absl/base/thread_annotations.h>
#include <absl/container/flat_hash_map.h>
#include <absl/container/flat_hash_set.h>
#include <absl/status/status.h>
#include <absl/status/statusor.h>
#include <absl/strings/str_format.h>
#include <absl/time/time.h>
 
#include "a11/actions/log.h"
#include "a11/actions/schema.h"
#include "a11/concurrency/future.h"
#include "a11/data/serialization.h"
#include "a11/data/types.h"
#include "a11/obs/span.h"
#include "thread/boost_primitives.h"
 
namespace thread {
class PermanentEvent;
}  // namespace thread
 
namespace a11::net {
class WireStream;
}  // namespace a11::net
 
namespace a11::nodes {
class AsyncNode;
class NodeMap;
}  // namespace a11::nodes
 
namespace a11::service {
class Session;
}  // namespace a11::service
 
namespace a11::actions {
 
class Action;
class ActionRegistry;
struct VerifiedAuthorization;
 
inline constexpr size_t kDefaultMaxConcurrentNestedActions = 64;
 
using ActionHandler = std::function<a11::Task(std::shared_ptr<Action>)>;
using SyncActionHandler = std::function<absl::Status(std::shared_ptr<Action>)>;
using OnActionCancelled = std::function<absl::Status(std::shared_ptr<Action>)>;
 
ActionHandler MakeAsyncActionHandler(SyncActionHandler handler);
 
class ActionLimiter {
 public:
  static absl::StatusOr<std::shared_ptr<ActionLimiter>> Create(size_t maximum);
  bool TryAcquire();
  absl::Status Acquire();
  void Release();
 
 private:
  explicit ActionLimiter(size_t maximum);
 
  thread::Mutex mu_;
  const size_t maximum_;
  size_t active_ ABSL_GUARDED_BY(mu_) = 0;
  std::shared_ptr<thread::PermanentEvent> changed_ ABSL_GUARDED_BY(mu_);
};
 
class Action : public std::enable_shared_from_this<Action> {
 public:
  static absl::StatusOr<std::shared_ptr<Action>> Create(
      ActionSchema schema, std::string action_id = {},
      ActionHandler handler = {},
      std::shared_ptr<nodes::NodeMap> node_map = nullptr,
      std::shared_ptr<net::WireStream> stream = nullptr,
      std::shared_ptr<service::Session> session = nullptr,
      std::shared_ptr<ActionRegistry> registry = nullptr,
      size_t max_concurrent_nested_actions =
          kDefaultMaxConcurrentNestedActions);
 
  static absl::StatusOr<std::string> MakeNodeId(std::string_view action_id,
                                                std::string_view node_name);
 
  [[nodiscard]] std::string GetId() const;
  absl::Status SetId(std::string action_id);
  [[nodiscard]] ActionSchema GetSchema() const;
  absl::Status SetSchema(ActionSchema schema);
  absl::Status BindHandler(ActionHandler handler);
  [[nodiscard]] ActionHandler GetHandler() const;
  [[nodiscard]] bool HasHandler() const;
 
  [[nodiscard]] ActionSettings GetSettings() const;
  absl::Status SetSettings(ActionSettings settings);
  absl::Status BindStreamsOnInputsByDefault(bool bind);
  absl::Status BindStreamsOnOutputsByDefault(bool bind);
  absl::Status ClearInputsAfterRun(bool clear = true);
  absl::Status ClearOutputsAfterRun(bool clear = true);
 
  absl::Status BindNodeMap(std::shared_ptr<nodes::NodeMap> node_map);
  [[nodiscard]] std::shared_ptr<nodes::NodeMap> GetNodeMap() const;
  absl::Status BindStream(std::shared_ptr<net::WireStream> stream);
  [[nodiscard]] std::shared_ptr<net::WireStream> GetStream() const;
  absl::Status BindRegistry(std::shared_ptr<ActionRegistry> registry);
  [[nodiscard]] std::shared_ptr<ActionRegistry> GetRegistry() const;
  absl::Status BindSession(const std::shared_ptr<service::Session>& session);
  [[nodiscard]] std::shared_ptr<service::Session> GetSession() const;
  absl::Status BindVerifiedAuthorization(
      std::shared_ptr<const VerifiedAuthorization> authorization);
  [[nodiscard]] std::shared_ptr<const VerifiedAuthorization>
  GetVerifiedAuthorization() const;
 
  absl::StatusOr<std::shared_ptr<nodes::AsyncNode>> GetNode(
      std::string node_id);
  absl::StatusOr<std::shared_ptr<nodes::AsyncNode>> GetInput(
      const std::string& name, std::optional<bool> bind_stream = std::nullopt);
  absl::StatusOr<std::shared_ptr<nodes::AsyncNode>> GetOutput(
      const std::string& name, std::optional<bool> bind_stream = std::nullopt);
  absl::StatusOr<std::shared_ptr<nodes::AsyncNode>> GetPort(
      const std::string& name);
  [[nodiscard]] bool ContainsPort(std::string_view name) const;
 
  template <typename T>
  absl::Status Log(const T& value, const LogOptions& options = {});
  absl::Status Log(data::Chunk chunk, const LogOptions& options = {});
  template <typename... Args>
  absl::Status Logf(const absl::FormatSpec<Args...>& format,
                    const Args&... args);
  template <typename... Args>
  absl::Status LogfWith(const LogOptions& options,
                        const absl::FormatSpec<Args...>& format,
                        const Args&... args);
  absl::StatusOr<std::shared_ptr<nodes::AsyncNode>> GetLogNode();
 
  [[nodiscard]] data::ActionMessage GetActionMessage() const;
  absl::Status MapPortsFromMessage(const data::ActionMessage& message);
 
  [[nodiscard]] data::ByteMap Headers() const;
  absl::StatusOr<std::optional<data::Bytes>> GetHeader(
      std::string_view name) const;
  [[nodiscard]] bool HasHeader(std::string_view name) const;
  absl::Status SetHeader(const std::string& name, data::Bytes value);
  absl::Status RemoveHeader(std::string_view name);
  absl::Status ForwardHeader(const std::shared_ptr<Action>& target,
                             std::string_view name) const;
  absl::Status ForwardHeadersWithPrefix(
      const std::shared_ptr<Action>& target,
      std::string_view prefix = kActionHeaderPrefix) const;
 
  absl::StatusOr<std::shared_ptr<Action>> MakeNested(
      const ActionSchema& schema, bool propagate_io = true,
      bool forward_headers = true);
  absl::StatusOr<std::shared_ptr<Action>> MakeNested(
      std::string_view action_name, bool propagate_io = true,
      bool forward_headers = true);
 
  absl::StatusOr<std::shared_ptr<Action>> Run();
  a11::Future<std::shared_ptr<Action>> Call(data::ByteMap wire_headers = {});
  a11::Future<absl::Status> WaitForDispatch(
      absl::Duration timeout = absl::InfiniteDuration());
  a11::Future<std::shared_ptr<Action>> Wait(
      absl::Duration timeout = absl::InfiniteDuration());
  absl::Status Cancel();
  absl::Status SetOnCancelled(OnActionCancelled callback);
 
  [[nodiscard]] std::string TraceId() const;
  [[nodiscard]] std::string SpanId() const;
 
  void SetSpanAttribute(std::string_view key, std::string_view value);
  void SetSpanAttribute(std::string_view key, std::int64_t value);
  void SetSpanAttribute(std::string_view key, bool value);
  void SetSpanAttribute(std::string_view key, double value);
  void SetSpanName(std::string_view name);
  void SetSpanStatus(obs::SpanStatus status, std::string_view description = {});
 
  [[nodiscard]] bool IsDone() const;
  [[nodiscard]] bool HasBeenRun() const;
  [[nodiscard]] bool HasBeenCalled() const;
  [[nodiscard]] bool Cancelled() const;
  [[nodiscard]] absl::Status GetStatus() const;
  [[nodiscard]] std::optional<absl::Status> GetDispatchStatus() const;
 
 private:
  enum class Mode { kNone, kRun, kCall, kCancelled };
 
  Action(ActionSchema schema, std::string id, ActionHandler handler,
         std::shared_ptr<nodes::NodeMap> node_map,
         std::shared_ptr<net::WireStream> stream,
         const std::shared_ptr<service::Session>& session,
         std::shared_ptr<ActionRegistry> registry,
         std::shared_ptr<ActionLimiter> nested_limiter);
 
  absl::Status Begin(Mode mode);
  absl::Status RemapDefaultPorts() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mu_);
  absl::Status AttachStreamIfRequested(
      const std::shared_ptr<nodes::AsyncNode>& node, bool bind);
  absl::Status ValidateMessagePorts(
      const std::vector<data::Port>& ports,
      const absl::flat_hash_map<std::string, ActionPortSchema>& schema_ports,
      std::string_view kind) const;
  void RunHandler(const std::shared_ptr<ActionLimiter>& limiter);
  absl::Status ApplyInputAutofills();
  [[nodiscard]] std::vector<data::NodeFragment> CollectAutofillFragments()
      const;
  void StartFinish(absl::Status status);
  absl::Status FinishRun(absl::Status status);
  // Whether any input port declares autofills, which are written before the
  // handler runs and so must not happen on the caller's thread.
  [[nodiscard]] bool HasInputAutofills() const;
  // Starts the handler on the calling thread and continues from its task
  // without a fiber. Only valid when nothing before the handler can block.
  absl::StatusOr<std::shared_ptr<Action>> RunHandlerWithoutFiber(
      const std::shared_ptr<Action>& self,
      std::shared_ptr<ActionLimiter> acquired_limiter = nullptr);
  void ReleaseAcquiredLimiter();
  absl::Status FinishOutputNodes(const absl::Status& status);
  // Applies an already-finished Action's terminal state to an output node that
  // is only being materialised now, so a late reader sees the end of the
  // stream.
  static absl::Status CloseUnwrittenOutput(
      const std::shared_ptr<nodes::AsyncNode>& node,
      const absl::Status& status);
  // Applies `options` to `chunk`, reports it to the process sink, and writes it
  // to the log port when anything could read it. The one place the log policy
  // lives; the Log/Logf templates only build the chunk.
  absl::Status WriteLog(data::Chunk chunk, const LogOptions& options);
  absl::Status CommunicateStatus(const absl::Status& status);
  absl::Status AbortInputs(const absl::Status& status);
  absl::Status SendNodeAbortStatuses(
      const absl::flat_hash_set<std::string>& node_ids,
      const absl::Status& status);
  absl::Status ReleaseNodesAfterRun();
  absl::Status DetachBoundStreamNodes();
  absl::Status SendRemoteCancel();
  void CompleteCall(const absl::Status& status, bool remove_from_session);
  void AbortLocalCallOutputs(const absl::Status& status);
  absl::Status TrackInSession(const std::shared_ptr<service::Session>& session);
  void UntrackFromSession();
  void SetDispatchStatus(absl::Status status);
  void SetCompletionStatus(const absl::Status& status);
 
  // Tracing hooks (a11::obs).
  absl::Status StartActionSpan(Mode mode);
  void EndActionSpan(const absl::Status& status);
  obs::Span MakeChildSpan(std::string_view name, obs::SpanKind kind);
  void RecordActionCallEvent(std::string_view name, std::string_view id);
 
  mutable thread::Mutex mu_;
  ActionSchema schema_ ABSL_GUARDED_BY(mu_);
  ActionHandler handler_ ABSL_GUARDED_BY(mu_);
  std::string id_ ABSL_GUARDED_BY(mu_);
  data::ByteMap headers_ ABSL_GUARDED_BY(mu_);
  ActionSettings settings_ ABSL_GUARDED_BY(mu_);
  std::shared_ptr<nodes::NodeMap> node_map_ ABSL_GUARDED_BY(mu_);
  std::shared_ptr<net::WireStream> stream_ ABSL_GUARDED_BY(mu_);
  std::weak_ptr<service::Session> session_ ABSL_GUARDED_BY(mu_);
  std::weak_ptr<service::Session> tracked_session_ ABSL_GUARDED_BY(mu_);
  std::shared_ptr<ActionRegistry> registry_ ABSL_GUARDED_BY(mu_);
  std::shared_ptr<const VerifiedAuthorization> verified_authorization_
      ABSL_GUARDED_BY(mu_);
  absl::flat_hash_map<std::string, std::string> input_ids_ ABSL_GUARDED_BY(mu_);
  absl::flat_hash_map<std::string, std::string> output_ids_
      ABSL_GUARDED_BY(mu_);
  absl::flat_hash_set<std::shared_ptr<nodes::AsyncNode>> input_nodes_
      ABSL_GUARDED_BY(mu_);
  absl::flat_hash_set<std::shared_ptr<nodes::AsyncNode>> output_nodes_
      ABSL_GUARDED_BY(mu_);
  absl::flat_hash_set<std::shared_ptr<nodes::AsyncNode>> stream_bound_nodes_
      ABSL_GUARDED_BY(mu_);
  Mode mode_ ABSL_GUARDED_BY(mu_) = Mode::kNone;
  bool input_autofills_applied_ ABSL_GUARDED_BY(mu_) = false;
  a11::Task task_ ABSL_GUARDED_BY(mu_);
  std::shared_ptr<ActionLimiter> acquired_limiter_ ABSL_GUARDED_BY(mu_);
  obs::Span span_ ABSL_GUARDED_BY(mu_);
  bool span_status_set_by_user_ ABSL_GUARDED_BY(mu_) = false;
  bool cancel_requested_ ABSL_GUARDED_BY(mu_) = false;
  bool finishing_ ABSL_GUARDED_BY(mu_) = false;
  std::optional<absl::Status> completion_status_ ABSL_GUARDED_BY(mu_);
  // Set once the Action has closed its output ports, with the status it closed
  // them under. Read by GetOutput() so a port materialised after that point
  // carries the same terminal state as one closed in place.
  bool outputs_finished_ ABSL_GUARDED_BY(mu_) = false;
  absl::Status outputs_final_status_ ABSL_GUARDED_BY(mu_);
  // Set once a consumer has taken the log port through GetLogNode(). It then
  // owns presentation, so WriteLog stops reporting to the process sink.
  bool log_claimed_ ABSL_GUARDED_BY(mu_) = false;
  std::optional<absl::Status> dispatch_status_ ABSL_GUARDED_BY(mu_);
  std::shared_ptr<a11::Promise<a11::Unit>> done_promise_ ABSL_GUARDED_BY(mu_);
  a11::Task done_future_ ABSL_GUARDED_BY(mu_);
  std::shared_ptr<a11::Promise<a11::Unit>> dispatch_promise_
      ABSL_GUARDED_BY(mu_);
  a11::Task dispatch_future_ ABSL_GUARDED_BY(mu_);
  std::vector<OnActionCancelled> cancel_callbacks_ ABSL_GUARDED_BY(mu_);
  std::weak_ptr<Action> parent_ ABSL_GUARDED_BY(mu_);
  absl::flat_hash_set<std::shared_ptr<Action>> children_ ABSL_GUARDED_BY(mu_);
  std::shared_ptr<ActionLimiter> nested_limiter_ ABSL_GUARDED_BY(mu_);
 
  friend class ActionRegistry;
  friend class service::Session;
};
 
namespace internal {
 
template <typename T>
inline constexpr bool kLogsAsText =
    std::is_convertible_v<const T&, std::string_view>;
 
}  // namespace internal
 
template <typename T>
absl::Status Action::Log(const T& value, const LogOptions& options) {
  const data::SerializationRegistry& registry =
      data::GlobalSerializationRegistry();
  absl::StatusOr<data::Chunk> chunk;
  if constexpr (internal::kLogsAsText<T>) {
    // Normalised to std::string first: the registry has a codec for that, not
    // for a pointer or an array of char.
    const std::string text{std::string_view(value)};
    chunk = registry.ToChunk<std::string>(text, options.mimetype.empty()
                                                    ? data::kTextMimetype
                                                    : options.mimetype);
  } else {
    chunk = registry.ToChunk<T>(value, options.mimetype);
  }
  if (!chunk.ok()) {
    return chunk.status();
  }
  return WriteLog(*std::move(chunk), options);
}
 
template <typename... Args>
absl::Status Action::Logf(const absl::FormatSpec<Args...>& format,
                          const Args&... args) {
  return Log(absl::StrFormat(format, args...), LogOptions{});
}
 
template <typename... Args>
absl::Status Action::LogfWith(const LogOptions& options,
                              const absl::FormatSpec<Args...>& format,
                              const Args&... args) {
  return Log(absl::StrFormat(format, args...), options);
}
 
}  // namespace a11::actions
 
#endif  // A11_ACTIONS_ACTION_H_