partisan_hyparview_peer_service_manager (partisan v6.0.0)

View Source

HyParView membership — a reliable partial-view overlay for gossip that stays connected under high node-failure rates (Leitão, Pereira & Rodrigues, HyParView: a Membership Protocol for Reliable Gossip-Based Broadcast, DSN 2007). This module implements the partisan_peer_service_manager behaviour.

The problem it solves

Gossip and epidemic broadcast need every node to know a set of peers to talk to. A full membership view does not scale; a random partial view can silently split the cluster into disconnected components. HyParView keeps two partial views of different sizes and roles so the overlay stays a single connected graph, with high probability, even as many nodes fail at once. The failure detector is just TCP: a dropped connection is treated as a failed peer, which makes it unreliable — a congestion spike can look like a failure — so the whole design is built to tolerate false positives.

The two views

  • Active view (active_max_size, default 6) — a small, symmetric set of peers this node holds a live connection to. The active views of all nodes, together, form the connected dissemination overlay that broadcast forwards over. Symmetric means if A holds B active then B holds A — the invariant the overlay's connectivity rests on.
  • Passive view (passive_max_size, default 30) — a larger set of backup peers, not connected, kept fresh so a failed active peer can be replaced at once. When the active view has a free slot the node promotes a passive peer with a neighbor request; a neighbor_request carries a priority, so a node with an empty active view can insist.

The views are probabilistic: the protocol does not prevent a partition, it makes one unlikely and repairs it reactively.

Joining and view construction

  • join — a new node contacts a contact node, which adds it to its active view and starts a forward_join random walk. The walk carries the active random-walk length (ARWL) as a TTL; as it is forwarded hop by hop with decreasing TTL, nodes add the joiner to their active view, and at the passive random-walk length (PRWL) to their passive view — spreading knowledge of the joiner across the overlay rather than clustering it at the contact node.
  • neighbor / neighbor_request — promote a passive peer into a free active slot.
  • disconnect — sent when a node leaves or drops a peer, so the other side moves it from active to passive.

Passive-view maintenance: shuffle

Periodically (shuffle_interval, default 10 s) a node runs a shuffle with a random active peer: it sends a sample of its own identity plus shuffle_k_active active and shuffle_k_passive passive peers; the peer merges them into its passive view and answers with a shuffle_reply sample of its own. This keeps every node's passive view a fresh, well-mixed sample of the cluster, so replacements for failed active peers are current rather than stale.

Failure handling and self-healing

A dropped connection removes a peer from the active view; the node immediately promotes a passive peer to refill the slot, holding the active view at size and the overlay connected. Because the detector is unreliable, a wrongly-dropped peer is simply re-added later — correctness never depends on the detector being right.

X-BOT — optimising the overlay

On top of plain HyParView this module implements X-BOT (Leitão, Marques, Pereira & Rodrigues, X-BOT: A Protocol for Resilient Optimization of Unstructured Overlay Networks): a node periodically tries to swap an active-view link for a better one — by a configurable cost such as latency — through a four-node handshake that preserves both connectivity and the active-view size throughout. It improves the overlay's shape without weakening the resilience HyParView guarantees.

Partisan additions to the paper

  • Active-view symmetry maintenance. Each node periodically re-asserts its active membership to its active peers with an ordinary neighbor message: a peer missing this node re-adds it, one that already holds it ignores it. This repairs a stable one-sided active view left when a control message is lost during churn, using no new wire message. Its cadence is active_view_maintenance_interval, which defaults to random_promotion_interval.
  • Epochs. epoch counts this node's restarts and epoch_count counts the disconnect messages it has generated; together they form a message_id() :: {epoch, epoch_count} that lets a node ignore stale control messages from a previous incarnation of a peer.
  • Partition injection / resolution. inject_partition and resolve_partition sever and restore overlay links deterministically, for tests and operations.

Configuration

active_max_size (6), passive_max_size (30), random_promotion and random_promotion_interval, shuffle_interval (10 s), shuffle_k_active (3), shuffle_k_passive (4), and the active/passive random-walk lengths ARWL and PRWL. Larger views trade memory and maintenance traffic for resilience.

Reading guide

init/1 builds the empty views and starts the shuffle and promotion timers. Membership operations arrive through the partisan_peer_service_manager callbacks; the protocol itself lives in the handlers for join, forward_join, neighbor, disconnect and shuffle, with the X-BOT optimisation and the peer-set bookkeeping (add-to-active, add-to-passive, promotion, symmetry maintenance) below them.

Summary

Types

active()

-type active() :: sets:set(partisan:node_spec()).

call()

-type call() ::
          {join, partisan:node_spec()} |
          {leave, partisan:node_spec()} |
          {update_members, [partisan:node_spec()]} |
          {resolve_partition, reference()} |
          {inject_partition, partisan:node_spec(), integer()} |
          {reserve, tag()} |
          active | passive |
          {active, tag()} |
          {send_message, node(), term()} |
          members | members_for_orchestration | get_local_state | connections | partitions.

cast()

-type cast() ::
          {join, partisan:node_spec()} |
          {receive_message, partisan:node_spec(), partisan:channel(), term()} |
          {disconnect, partisan:node_spec()}.

config()

-type config() ::
          #{active_max_size := non_neg_integer(),
            active_min_size := non_neg_integer(),
            active_rwl := non_neg_integer(),
            passive_max_size := non_neg_integer(),
            passive_rwl := non_neg_integer(),
            random_promotion := boolean(),
            random_promotion_interval := non_neg_integer(),
            shuffle_interval := non_neg_integer(),
            shuffle_k_active := non_neg_integer(),
            shuffle_k_passive := non_neg_integer(),
            xbot_enabled := boolean(),
            xbot_interval := non_neg_integer()}.

epoch()

-type epoch() :: non_neg_integer().

epoch_count()

-type epoch_count() :: non_neg_integer().

message_id()

-type message_id() :: {epoch(), epoch_count()}.

message_id_store()

-type message_id_store() :: #{partisan:node_spec() := message_id()}.

passive()

-type passive() :: sets:set(partisan:node_spec()).

reserved()

-type reserved() :: #{atom() := partisan:node_spec()}.

t()

-type t() ::
          #state{name :: node(),
                 node_spec :: partisan:node_spec(),
                 config :: config(),
                 active :: active(),
                 passive :: passive(),
                 reserved :: reserved(),
                 out_links :: list(),
                 tag :: tag(),
                 epoch :: epoch(),
                 sent_message_map :: message_id_store(),
                 recv_message_map :: message_id_store(),
                 partitions :: partisan_peer_service_manager:partitions()}.

tag()

-type tag() :: atom().

Functions

active()

active(Tag)

cast_message(Term, Message)

-spec cast_message(Term :: partisan:any_pid() | partisan:any_name(), Message :: partisan:message()) ->
                      ok.

cast_message(Node, ServerRef, Message)

cast_message(Node, ServerRef, Message, Options)

code_change(OldVsn, State, Extra)

-spec code_change(term() | {down, term()}, t(), term()) -> {ok, t()}.

decode/1

-spec decode({state, sets:set(), any()} | sets:set()) -> list().

delete_state_from_disk()

forward_message(Term, Message)

forward_message/3

forward_message/4

get_local_state()

-spec get_local_state() -> {state, Active :: active(), Epoch :: integer()}.

handle_call/3

-spec handle_call(call(), {pid(), term()}, t()) -> {reply, term(), t()}.

handle_cast/2

-spec handle_cast(cast(), t()) -> {noreply, t()}.

handle_info/2

-spec handle_info(term(), t()) -> {noreply, t()}.

init/1

-spec init([]) -> {ok, t()} | {stop, reservation_limit_exceeded}.

inject_partition(Origin, TTL)

join(Node)

leave()

leave(Node)

members()

-spec members() -> [node()].

members_for_orchestration()

-spec members_for_orchestration() -> [partisan:node_spec()].

on_down(Name, Function)

on_down(Name, Function, Opts)

on_up(Name, Function)

on_up(Name, Function, Opts)

partitions()

passive()

receive_message/3

reserve(Tag)

resolve_partition(Reference)

send_message(Name, Message)

start_link()

-spec start_link() -> {ok, pid()} | ignore | {error, term()}.

supports_capability(Arg)

-spec supports_capability(Arg :: atom()) -> boolean().

sync_join(Node)

terminate(Reason, State)

-spec terminate(term(), t()) -> term().

update_members(Members)