View Source ProcessHub.Service.RequestManager (ProcessHub v0.5.0-beta)

Unified module for distributed operation tracking and execution utilities.

This module provides:

  • Operation struct for tracking distributed operations across nodes
  • State management (store, get, update, remove from coordinator state)
  • GenServer handlers for coordinator callbacks
  • Execution utilities (partition check, send response, request factories)
  • Tracking utilities (expired?, all_responded?, record_response)
  • Request splitting for batching large requests

Summary

Types

t()

Functions

all_responded?(map)
cleanup_expired(state)
compose_sub_requests(operation)

Composes sub-requests for each target node.

contraction_request(hub, children_data, opts \\ [migration_add: true])

Factory: Creates a contraction request for topology contraction.

expired?(map)
finalize(operation, hub)

Finalizes an operation by aggregating results and running post-processing.

get(state, transaction_id)
handle_await(state, transaction_id, from)
handle_response(state, transaction_id, response_node, results)
handle_timeout(state, transaction_id, from)
load_strategies(hub)

Loads all strategies from storage.

migration_request(hub, target_node, children_data, opts \\ [])

Factory: Creates a migration request for topology expansion.

new(hub, handler_module, nodes_data, opts)

Creates a new Operation struct.

populate_forward(forward_data, target_nodes, child_data)

Groups children by target node into a keyword list [{node, [child_data, ...]}, ...].

process_response(operation, response_node, results)

Records a node's response and checks if operation is complete.

record_response(request, response_node, results, results_field)
remove(state, transaction_id)
request_to_opts(request)

Converts request struct to keyword options for response routing.

send_response(response_type, opts, results)

Sends response to coordinator.

set_awaiter(request, from)
split(req)

Splits large requests into smaller batches for efficient cross-node transmission.

store(state, operation)
update(state, operation)
with_partition_check(hub, fun)

Wraps function execution with partition check.

Types

Link to this type

t()

View Source 
@type t() :: %ProcessHub.Service.RequestManager{
  awaiter: {pid(), reference()} | nil,
  completed_nodes: MapSet.t(),
  expires_at: integer(),
  future: ProcessHub.Future.t() | nil,
  handler: module(),
  hub_id: ProcessHub.hub_id(),
  nodes_data: [{node(), [map()]}],
  options: keyword(),
  sub_requests: [struct()],
  transaction_id: reference()
}

Functions

Link to this function

all_responded?(map)

View Source 
@spec all_responded?(map()) :: boolean()
Link to this function

cleanup_expired(state)

View Source 
@spec cleanup_expired(ProcessHub.Hub.t()) :: ProcessHub.Hub.t()
Link to this function

compose_sub_requests(operation)

View Source 
@spec compose_sub_requests(t()) :: {:ok, t()} | {:error, :no_children}

Composes sub-requests for each target node.

Link to this function

contraction_request(hub, children_data, opts \\ [migration_add: true])

View Source 
@spec contraction_request(
  ProcessHub.Hub.t(),
  [{ProcessHub.child_spec(), map()}],
  keyword()
) ::
  ProcessHub.Request.Handler.StartChildrenRequest.t()

Factory: Creates a contraction request for topology contraction.

Link to this function

expired?(map)

View Source 
@spec expired?(map()) :: boolean()
Link to this function

finalize(operation, hub)

View Source 
@spec finalize(t(), ProcessHub.Hub.t()) :: struct()

Finalizes an operation by aggregating results and running post-processing.

Link to this function

get(state, transaction_id)

View Source 
@spec get(ProcessHub.Hub.t(), reference()) :: t() | nil
Link to this function

handle_await(state, transaction_id, from)

View Source 
@spec handle_await(ProcessHub.Hub.t(), reference(), {pid(), reference()}) ::
  {:reply, term(), ProcessHub.Hub.t()} | {:noreply, ProcessHub.Hub.t()}
Link to this function

handle_response(state, transaction_id, response_node, results)

View Source 
@spec handle_response(ProcessHub.Hub.t(), reference(), node(), term()) ::
  {:noreply, ProcessHub.Hub.t()}
Link to this function

handle_timeout(state, transaction_id, from)

View Source 
@spec handle_timeout(ProcessHub.Hub.t(), reference(), {pid(), reference()}) ::
  {:noreply, ProcessHub.Hub.t()}
Link to this function

load_strategies(hub)

View Source 
@spec load_strategies(ProcessHub.Hub.t()) :: %{
  sync: term(),
  dist: term(),
  redun: term(),
  migr: term()
}

Loads all strategies from storage.

Link to this function

migration_request(hub, target_node, children_data, opts \\ [])

View Source 
@spec migration_request(
  ProcessHub.Hub.t(),
  node(),
  [{ProcessHub.child_spec(), map()}],
  keyword()
) ::
  ProcessHub.Request.Handler.StartChildrenRequest.t()

Factory: Creates a migration request for topology expansion.

Link to this function

new(hub, handler_module, nodes_data, opts)

View Source 
@spec new(ProcessHub.Hub.t(), module(), [{node(), [map()]}], keyword()) :: t()

Creates a new Operation struct.

Link to this function

populate_forward(forward_data, target_nodes, child_data)

View Source 
@spec populate_forward(keyword(), [node()], map()) :: keyword()

Groups children by target node into a keyword list [{node, [child_data, ...]}, ...].

Link to this function

process_response(operation, response_node, results)

View Source 
@spec process_response(t(), node(), term()) :: {:complete, t()} | {:pending, t()}

Records a node's response and checks if operation is complete.

Link to this function

record_response(request, response_node, results, results_field)

View Source 
@spec record_response(map(), node(), term(), atom()) :: map()
Link to this function

remove(state, transaction_id)

View Source 
@spec remove(ProcessHub.Hub.t(), reference()) :: ProcessHub.Hub.t()
Link to this function

request_to_opts(request)

View Source 
@spec request_to_opts(map()) :: keyword()

Converts request struct to keyword options for response routing.

Link to this function

send_response(response_type, opts, results)

View Source 
@spec send_response(atom(), keyword(), [{ProcessHub.child_id(), term()}]) ::
  :ok | :skip

Sends response to coordinator.

Link to this function

set_awaiter(request, from)

View Source 
@spec set_awaiter(
  map(),
  {pid(), reference()}
) :: map()
Link to this function

split(req)

View Source 
@spec split(struct()) :: [struct()]

Splits large requests into smaller batches for efficient cross-node transmission.

Link to this function

store(state, operation)

View Source 
@spec store(ProcessHub.Hub.t(), t()) :: ProcessHub.Hub.t()
Link to this function

update(state, operation)

View Source 
@spec update(ProcessHub.Hub.t(), t()) :: ProcessHub.Hub.t()
Link to this function

with_partition_check(hub, fun)

View Source 
@spec with_partition_check(ProcessHub.Hub.t(), (-> term())) ::
  term() | {:error, :partitioned}

Wraps function execution with partition check.