Worker (octez-libs.Tezos_gossipsub.Worker)

Parameters

module C : Gossipsub_intf.WORKER_CONFIGURATION

Signature

type t

The state of a gossipsub worker.

exception Worker_crashed

We (re-)export the GS, Monad and Stream modules.

include Gossipsub_intf.WORKER_CONFIGURATION
  with module GS = C.GS
  with module GS.Topic = C.GS.Topic
  with module GS.Peer = C.GS.Peer
  with module Monad = C.Monad
  with module Stream = C.Stream
  with module Point = C.Point

module GS : 
  module type of struct include C.GS end
    with module Topic = C.GS.Topic
    with module Peer = C.GS.Peer

The gossipsub automaton that will be used by the worker.

module Point = C.Point

val maybe_reachable_point : GS.Peer.t -> Point.t

module Monad = C.Monad

Abstraction of the IO monad used by the worker.

module Stream = C.Stream

A mutable (FIFO) stream of data.

type message_with_header = {

message : GS.Message.t;
topic : GS.Topic.t;
message_id : GS.Message_id.t;

}

A message together with a header, that is, a topic and an id. This corresponds to what the spec calls a "full message".

TODO: https://gitlab.com/tezos/tezos/-/issues/5415 The topic can be inferred from the message_id and can thus be omitted; this also applies for the IHave message.

type p2p_message =

| Graft of {
1. topic : GS.Topic.t;
}
| Prune of {
1. topic : GS.Topic.t;
2. px : GS.Peer.t Tezos_base.TzPervasives.Seq.t;
3. backoff : GS.Span.t;
}
| IHave of {
1. topic : GS.Topic.t;
2. message_ids : GS.Message_id.t list;
}
| IWant of {
1. message_ids : GS.Message_id.t list;
}
| Subscribe of {
1. topic : GS.Topic.t;
}
| Unsubscribe of {
1. topic : GS.Topic.t;
}
| Message_with_header of message_with_header
| Ping

The following type defines the different kinds of messages a peer could receive from or sent to the P2P layer.

type p2p_input =

| In_message of {
1. from_peer : GS.Peer.t;
2. p2p_message : p2p_message;
}
| New_connection of {
1. peer : GS.Peer.t;
2. direct : bool;
3. trusted : bool;
4. bootstrap : bool;
}
| Disconnection of {
1. peer : GS.Peer.t;
}

The different kinds of input events that could be received from the P2P layer.

type app_input =

| Publish_message of message_with_header
| Join of GS.Topic.t
| Leave of GS.Topic.t

The different kinds of input events that could be received from the application layer.

type peer_origin =

| PX of GS.Peer.t
| Trusted

A peer's origin is either another peer (i.e. advertised via PX), or none if it is trusted.

type p2p_output =

| Out_message of {
1. to_peer : GS.Peer.t;
2. p2p_message : p2p_message;
}
(*
Emit the given p2p_message to the remote peer to_peer.
*)
| Disconnect of {
1. peer : GS.Peer.t;
}
(*
End the connection with the peer peer.
*)
| Kick of {
1. peer : GS.Peer.t;
}
(*
Kick the peer peer: the peer is disconnected and blacklisted.
*)
| Connect of {
1. peer : GS.Peer.t;
2. origin : peer_origin;
}
(*
Inform the p2p_output messages processor that we want to connect to the peer peer advertised by some other peer origin.
*)
| Connect_point of {
1. point : Point.t;
}
(*
Version of connect where we provide a point directly.
*)
| Forget of {
1. peer : GS.Peer.t;
2. origin : GS.Peer.t;
}
(*
Inform the p2p_output messages processor that we don't want to connect to the peer peer advertised by some other peer origin.
*)

The different kinds of outputs that could be emitted by the worker for the P2P layer.

type app_output = message_with_header

The application layer will be advertised about full messages it's interested in.

type event = private

| Heartbeat
| P2P_input of p2p_input
| App_input of app_input
| Check_unknown_messages
| Process_batch of (GS.receive_message * GS.Peer.Set.t) list

The different kinds of events the Gossipsub worker handles.

val make : 
  ?events_logging:(event -> unit Monad.t) ->
  ?initial_points:(unit -> Point.t list) ->
  ?batching_interval:GS.span ->
  self:GS.Peer.t ->
  Stdlib.Random.State.t ->
  (GS.Topic.t, GS.Peer.t, GS.Message_id.t, GS.span) Gossipsub_intf.limits ->
  (GS.Peer.t, GS.Message_id.t) Gossipsub_intf.parameters ->
  t

make ~events_logging ~initial_points rng limits parameters initializes a new Gossipsub automaton with the given arguments. Then, it initializes and returns a worker for it. The events_logging function can be used to define a handler for logging the worker's events. initial_points allows to resolve a list of known peers' addresses to which we may want to reconnect in the worker.

val start : GS.Topic.t list -> t -> unit

start topics state runs the (not already started) worker whose state is given together with the initial list of topics the caller is interested in.

val shutdown : t -> unit Monad.t

shutdown state allows stopping the worker whose state is given.

val app_input : t -> app_input -> unit

app_input state app_input adds the given application input app_input to the worker's input stream.

val set_unreachable_point : t -> Point.t -> unit

set_unreachable_point state point declares this point as unreachable.

val p2p_input : t -> p2p_input -> unit

p2p_input state p2p_input adds the given P2P input p2p_input to the worker's input stream.

val p2p_output_stream : t -> p2p_output Stream.t

p2p_output_stream t returns the output stream containing data for the P2P layer.

val app_output_stream : t -> app_output Stream.t

app_output_stream t returns the output stream containing data for the application layer.

val input_events_stream : t -> event Stream.t

input_events_stream t returns the input stream in which we push events to be processed by the worker.

val reconnection_delays : t -> (Point.t * GS.Span.t) list

reconnection_delays t returns the points that are currently unreachable with the span before the next attempt to reconnect to them.

val is_subscribed : t -> GS.Topic.t -> bool

is_subscribed t topic checks whether topic is in the mesh of t.

val pp_p2p_output : Stdlib.Format.formatter -> p2p_output -> unit

Pretty-printer for values of type p2p_output.

val pp_app_output : Stdlib.Format.formatter -> app_output -> unit

Pretty-printer for values of type app_output.

module Introspection : sig ... end

Introspection and stats facilities

val stats : t -> Introspection.stats

val state : t -> GS.Introspection.view

val worker_crashed : t -> unit Lwt.t