Libp2p.Gossipsub (libp2p_elixir v0.9.0)

Implements the Gossipsub v1.1 PubSub protocol.

Gossipsub is a scalable, extensible PubSub protocol that uses a mesh for efficient data dissemination and gossip for robustness. This module implements the v1.1 specification, which adds significant security and performance improvements over v1.0.

Key Features (v1.1)

  • Explicit Peering: Supports direct, persistent peering agreements.
  • Prune Backoff: When pruning a peer, a backoff time is enforced to prevent rapid re-grafting.
  • Peer Exchange (PX): Prune messages can contain a list of alternative peers to help bootstrapping without a DHT.
  • Flood Publishing: New messages from the self node are published to all peers (not just the mesh) to counter eclipse attacks.
  • Adaptive Gossip: Gossip emission targets a randomized factor of peers (0.25 default).
  • Outbound Mesh Quotas: Maintains a minimum number of outbound connections in the mesh to prevent Sybil attacks.

Scoring

While full peer scoring is defined in the spec, this implementation currently provides the structural support for scoring parameters (Time in Mesh, First Message Deliveries, Mesh Message Delivery Rate, etc.) to allow for future tuning/enforcement.

Core Mechanisms (common to v1.0 and v1.1)

  • Mesh Maintenance: Builds and maintains a mesh of peers for each topic.
  • Gossip: Disseminates message identifiers (IHAVE) to random peers to ensure propagation.
  • Control Messages: Handles GRAFT, PRUNE, IHAVE, and IWANT control messages.
  • Deduplication: Tracks seen message IDs to prevent re-propagation.

Limitations

  • Peer scoring is not yet fully implemented (only structural support).
  • Opportunistic grafting is not yet implemented.

Summary

Types

msg_id()
peer_id()
peer_state()
state()
topic()

Functions

await_peer(gossipsub, peer_id, timeout \\ 10000)

Block until an outbound gossipsub stream exists for peer_id.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

handle_inbound(conn, stream_id, initial)

Swarm stream router hook (3-arity), using default registered server name.

handle_inbound(gossipsub, conn, stream_id, initial)

Swarm stream router hook (explicit gossipsub process).

peer_connected(gossipsub, peer_id, conn)

Swarm hook: register that a peer connection exists.

publish(gossipsub, topic, data)
start_link(opts \\ [])
subscribe(gossipsub, topic)

Types

msg_id()

@type msg_id() :: binary()

peer_id()

@type peer_id() :: binary()

peer_state()

@type peer_state() :: %{
  conn: pid(),
  inbound_stream_id: non_neg_integer() | nil,
  outbound_stream_id: non_neg_integer() | nil,
  topics: MapSet.t(topic()),
  buf: binary()
}

state()

@type state() :: %{
  peers: %{required(peer_id()) => peer_state()},
  subscriptions: MapSet.t(topic()),
  mesh: %{required(topic()) => MapSet.t(peer_id())},
  seen: MapSet.t(msg_id()),
  mcache: %{required(msg_id()) => %{topic: topic(), msg: map()}},
  on_message: (topic(), binary(), peer_id() -> any()) | nil,
  msg_id_fn: (topic(), binary() -> msg_id()),
  peer_waiters: %{required(peer_id()) => [GenServer.from()]},
  event_sink: pid() | nil
}

topic()

@type topic() :: binary()

Functions

await_peer(gossipsub, peer_id, timeout \\ 10000)

@spec await_peer(pid() | atom(), peer_id(), timeout()) :: :ok | {:error, term()}

Block until an outbound gossipsub stream exists for peer_id.

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

handle_inbound(conn, stream_id, initial)

@spec handle_inbound(pid(), non_neg_integer(), binary()) :: :ok

Swarm stream router hook (3-arity), using default registered server name.

handle_inbound(gossipsub, conn, stream_id, initial)

@spec handle_inbound(pid() | atom(), pid(), non_neg_integer(), binary()) :: :ok

Swarm stream router hook (explicit gossipsub process).

peer_connected(gossipsub, peer_id, conn)

@spec peer_connected(pid() | atom(), peer_id(), pid()) :: :ok

Swarm hook: register that a peer connection exists.

publish(gossipsub, topic, data)

@spec publish(pid() | atom(), topic(), binary()) :: :ok

start_link(opts \\ [])

@spec start_link(keyword()) :: GenServer.on_start()

subscribe(gossipsub, topic)

@spec subscribe(pid() | atom(), topic()) :: :ok