View Source ProcessHub.Service.RequestManager (ProcessHub v0.8.0)

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

responder()

Who answers a sub-request: the node, or the node and the nodes that forwarded the request to it.

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.

forward_fields(hub, opts)

The fields of a request this node forwards on behalf of the request whose options are opts: the same transaction and originating node, so the target answers the coordinator that is waiting, and the forward path extended by this node, which names the answer and bounds the forwarding.

forward_loop?(nodes, path)

True when every node a child belongs to already forwarded it, so forwarding it again would only send it back.

forward_path(opts)

The nodes that forwarded the request whose options are opts, in order.

forwarded_answers(forward_data)

The answers a node gives for the children it forwarded, one {child_id, {:forwarded, node}} per target.

get(state, transaction_id)
handle_await(state, transaction_id, from)
handle_departed(state)

Answers each awaited operation that now waits only on nodes this node is no longer connected to. Their children answer :no_response at once, as they would at the timeout, so a caller can try again while the time it was given lasts. Called when a node goes down.

handle_response(state, transaction_id, responder, results)
handle_timeout(state, transaction_id, from)
load_strategies(hub)

Loads the strategies a child-start request resolves against.

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, responder, results)

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

record_response(request, responder, results, results_field)

Records the answer of responder to its sub-request.

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

responder()

View Source 
@type responder() :: node() | {node(), [node()]}

Who answers a sub-request: the node, or the node and the nodes that forwarded the request to it.

Link to this type

t()

View Source 
@type t() :: %ProcessHub.Service.RequestManager{
  awaiter: {pid(), reference()} | nil,
  expires_at: integer(),
  future: ProcessHub.Future.t() | nil,
  handler: module(),
  hub_id: ProcessHub.hub_id(),
  nodes_data: [{node(), [map()]}],
  options: keyword(),
  result: struct() | nil,
  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.Coordinator.State.t()) ::
  ProcessHub.Coordinator.State.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

forward_fields(hub, opts)

View Source 
@spec forward_fields(
  ProcessHub.Hub.t(),
  keyword()
) :: map()

The fields of a request this node forwards on behalf of the request whose options are opts: the same transaction and originating node, so the target answers the coordinator that is waiting, and the forward path extended by this node, which names the answer and bounds the forwarding.

Link to this function

forward_loop?(nodes, path)

View Source 
@spec forward_loop?([node()], [node()]) :: boolean()

True when every node a child belongs to already forwarded it, so forwarding it again would only send it back.

Link to this function

forward_path(opts)

View Source 
@spec forward_path(keyword()) :: [node()]

The nodes that forwarded the request whose options are opts, in order.

Link to this function

forwarded_answers(forward_data)

View Source 
@spec forwarded_answers(keyword()) :: [{ProcessHub.child_id(), {:forwarded, node()}}]

The answers a node gives for the children it forwarded, one {child_id, {:forwarded, node}} per target.

Link to this function

get(state, transaction_id)

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

handle_await(state, transaction_id, from)

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

handle_departed(state)

View Source 
@spec handle_departed(ProcessHub.Coordinator.State.t()) ::
  ProcessHub.Coordinator.State.t()

Answers each awaited operation that now waits only on nodes this node is no longer connected to. Their children answer :no_response at once, as they would at the timeout, so a caller can try again while the time it was given lasts. Called when a node goes down.

Link to this function

handle_response(state, transaction_id, responder, results)

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

handle_timeout(state, transaction_id, from)

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

load_strategies(hub)

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

Loads the strategies a child-start request resolves against.

Propagation reads the synchronization strategy itself, through ProcessHub.Service.Dispatcher.propagate_event/3.

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.

Each child carries migrated_from, the node it is moving away from.

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, responder, results)

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

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

Link to this function

record_response(request, responder, results, results_field)

View Source 
@spec record_response(map(), responder(), term(), atom()) :: map()

Records the answer of responder to its sub-request.

An answer naming a child as {:forwarded, node} adds a sub-request for that node's answer to the forward. A forward's answer can arrive before the answer announcing it; it is kept, and the announcement then finds it.

Link to this function

remove(state, transaction_id)

View Source 
@spec remove(ProcessHub.Coordinator.State.t(), reference()) ::
  ProcessHub.Coordinator.State.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.Coordinator.State.t(), t()) :: ProcessHub.Coordinator.State.t()
Link to this function

update(state, operation)

View Source 
@spec update(ProcessHub.Coordinator.State.t(), t()) ::
  ProcessHub.Coordinator.State.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.