Module Gossipsub.Worker

The worker module exposes instantiation of the Gossipsub worker functor, alongside the config used to instantiate the functor and the default values of the GS parameters.

This module defines the default values for simple Gossip parameters and limits. These values are used or re-defined in the DAL node before instantiating the Gossipsub worker.

include Mavryk_gossipsub.Gossipsub_intf.WORKER with type GS.Topic.t = Mavryk_dal_node_services.Types.Topic.t and type GS.Message_id.t = Mavryk_dal_node_services.Types.Message_id.t and type GS.Message.t = Mavryk_dal_node_services.Types.Message.t and type GS.Peer.t = Mavryk_dal_node_services.Types.Peer.t and type GS.Span.t = Mavryk_dal_node_services.Types.Span.t and type GS.Time.t = Mavryk_dal_node_services.Types.Time.t and type Point.t = Mavryk_dal_node_services.Types.Point.t
type t

The state of a gossipsub worker.

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

include Mavryk_gossipsub.Gossipsub_intf.WORKER_CONFIGURATION with type GS.Topic.t = Mavryk_dal_node_services.Types.Topic.t with type GS.Message_id.t = Mavryk_dal_node_services.Types.Message_id.t with type GS.Message.t = Mavryk_dal_node_services.Types.Message.t with type GS.Peer.t = Mavryk_dal_node_services.Types.Peer.t with type GS.Span.t = Mavryk_dal_node_services.Types.Span.t with type GS.Time.t = Mavryk_dal_node_services.Types.Time.t with type Point.t = Mavryk_dal_node_services.Types.Point.t

The gossipsub automaton that will be used by the worker.

module Monad : sig ... end

Abstraction of the IO monad used by the worker.

module Stream : sig ... end

A mutable (FIFO) stream of data.

type message_with_header = {
  1. message : GS.Message.t;
  2. topic : GS.Topic.t;
  3. 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".

type p2p_message =
  1. | Graft of {
    1. topic : GS.Topic.t;
    }
  2. | Prune of {
    1. topic : GS.Topic.t;
    2. px : GS.Peer.t Mavryk_base.TzPervasives.Seq.t;
    3. backoff : GS.Span.t;
    }
  3. | IHave of {
    1. topic : GS.Topic.t;
    2. message_ids : GS.Message_id.t list;
    }
  4. | IWant of {
    1. message_ids : GS.Message_id.t list;
    }
  5. | Subscribe of {
    1. topic : GS.Topic.t;
    }
  6. | Unsubscribe of {
    1. topic : GS.Topic.t;
    }
  7. | Message_with_header of message_with_header

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

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

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

type app_input =
  1. | Publish_message of message_with_header
  2. | Join of GS.Topic.t
  3. | Leave of GS.Topic.t

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

type peer_origin =
  1. | PX of GS.Peer.t
  2. | Trusted

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

type p2p_output =
  1. | 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.

    *)
  2. | Disconnect of {
    1. peer : GS.Peer.t;
    }
    (*

    End the connection with the peer peer.

    *)
  3. | Kick of {
    1. peer : GS.Peer.t;
    }
    (*

    Kick the peer peer: the peer is disconnected and blacklisted.

    *)
  4. | 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.

    *)
  5. | Connect_point of {
    1. point : Point.t;
    }
    (*

    Version of connect where we provide a point directly.

    *)
  6. | 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
  1. | Heartbeat
  2. | P2P_input of p2p_input
  3. | App_input of app_input

The different kinds of events the Gossipsub worker handles.

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

make ~events_logging ~bootstrap_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. The list of bootstrap_points represents the list of initially 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 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 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
module Logging : sig ... end
module Validate_message_hook : sig ... end

A hook to set or update messages and messages IDs validation function. Should be called once at startup and every time the DAL parameters change.