Source file gossipsub_intf.ml

(*****************************************************************************)
(*                                                                           *)
(* Open Source License                                                       *)
(* Copyright (c) 2023 Nomadic Labs, <contact@nomadic-labs.com>               *)
(* Copyright (c) 2023 Functori,     <contact@functori.com>                   *)
(*                                                                           *)
(* Permission is hereby granted, free of charge, to any person obtaining a   *)
(* copy of this software and associated documentation files (the "Software"),*)
(* to deal in the Software without restriction, including without limitation *)
(* the rights to use, copy, modify, merge, publish, distribute, sublicense,  *)
(* and/or sell copies of the Software, and to permit persons to whom the     *)
(* Software is furnished to do so, subject to the following conditions:      *)
(*                                                                           *)
(* The above copyright notice and this permission notice shall be included   *)
(* in all copies or substantial portions of the Software.                    *)
(*                                                                           *)
(* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR*)
(* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,  *)
(* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL   *)
(* THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER*)
(* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING   *)
(* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER       *)
(* DEALINGS IN THE SOFTWARE.                                                 *)
(*                                                                           *)
(*****************************************************************************)

module type ITERABLE = sig
  type t

  include COMPARABLE with type t := t

  include PRINTABLE with type t := t

  module Set : Set.S with type elt = t

  module Map : Map.S with type key = t
end

module type AUTOMATON_SUBCONFIG = sig
  module Peer : ITERABLE

  module Topic : ITERABLE

  module Message_id : sig
    include ITERABLE

    (** [get_topic id] returns/computes a message's topic from the given message
        id. Message ids should be defined so that this function can be
        implemented. *)
    val get_topic : t -> Topic.t
  end

  module Message : sig
    include PRINTABLE

    (** [valid] performs an application layer-level validity check on a message
        id and a message if given.

        The message id (and message if any) could either be [`Valid] in the
        current context or [`Invalid], meaning that it is/they are not valid (in
        the present time, in the past and in the future). The application layer
        could also return [`Outdated] or [`Unknown] if the message id is
        outdated or if the application doesn't care about validity. In this
        case, the application might omit some costly validity checks. *)
    val valid :
      ?message:t ->
      message_id:Message_id.t ->
      unit ->
      [`Valid | `Unknown | `Outdated | `Invalid]
  end
end

module type SPAN = sig
  include Compare.S

  include PRINTABLE with type t := t

  val zero : t

  val of_int_s : int -> t

  val to_int_s : t -> int

  val of_float_s : float -> t

  val to_float_s : t -> float

  (** [mul s n] returns [n * s]. *)
  val mul : t -> int -> t
end

module type TIME = sig
  include Compare.S

  include PRINTABLE with type t := t

  type span

  val now : unit -> t

  val add : t -> span -> t

  val sub : t -> span -> t

  val to_span : t -> span
end

module type AUTOMATON_CONFIG = sig
  module Subconfig : AUTOMATON_SUBCONFIG

  module Span : SPAN

  module Time : TIME with type span = Span.t
end

type 'span per_topic_score_limits = {
  time_in_mesh_weight : float;
      (** P1: The weight of the score associated to the time spent in the mesh. *)
  time_in_mesh_cap : float;
      (** P1: The maximum value considered for the score associated to the time spent
          by a peer in the mesh. *)
  time_in_mesh_quantum : 'span;
      (** P1: The score associated to the time spent [t] in the mesh is
          [(min t time_in_mesh_cap) / time_in_mesh_quantum]. *)
  first_message_deliveries_weight : float;
      (** P2: The weight of the score associated to the number of first message deliveries. *)
  first_message_deliveries_cap : int;
      (** P2: The maximum value considered during score computation for the number of
          first message deliveries. *)
  first_message_deliveries_decay : float;
      (** P2: The score is multiplied by this factor every [score_cleanup_ticks] heartbeat.
          This parameter must be in the unit interval. *)
  mesh_message_deliveries_weight : float;
      (** P3: The weight of the score associated to the number of first/near-first
          mesh message deliveries. *)
  mesh_message_deliveries_window : 'span;
      (** P3: The delay added to the first delivery of a message to obtain the upper bound up to which a
          duplicate message is counted as nearly-first delivered. *)
  mesh_message_deliveries_activation : 'span;
      (** P3: How long should a peer be in the mesh before we start evaluating P3. *)
  mesh_message_deliveries_cap : int;
      (** P3: The maximum value considered during score computation for the number of
          first/near-first mesh message deliveries. *)
  mesh_message_deliveries_threshold : int;
      (** P3: The number of messages received from a peer in the mesh in the
          associated topic above which the peer won't be penalized  *)
  mesh_message_deliveries_decay : float;
      (** P3: The score is multiplied by this factor every [score_cleanup_ticks] heartbeat.
          This parameter must be in the unit interval. *)
  mesh_failure_penalty_weight : float;
      (** P3b: Penalty induced when a peer gets pruned with a non-zero mesh message delivery deficit. *)
  mesh_failure_penalty_decay : float;
      (** P3b: The score is multiplied by this factor every [score_cleanup_ticks] heartbeat.
          This parameter must be in the unit interval. *)
  invalid_message_deliveries_weight : float;
      (** P4: Penalty induced when a peer sends an invalid message. *)
  invalid_message_deliveries_decay : float;
      (** P4: The score is multiplied by this factor every [score_cleanup_ticks] heartbeat.
          This parameter must be in the unit interval. *)
}

type ('topic, 'span) topic_score_limits =
  | Topic_score_limits_single of 'span per_topic_score_limits
      (** Use this constructor when the topic score parameters do not
          depend on the topic. *)
  | Topic_score_limits_family of {
      all_topics : 'topic Seq.t;
      parameters : 'topic -> 'span per_topic_score_limits;
      weights : 'topic -> float;
    }
      (** Use this constructor when the topic score parameters may depend
          on the topic. *)

(* N.B.: unlike the Go/Rust implementations, scores are not refreshed in an asynchronous
   loop but rather in the heartbeat. Hence, we do not have the [decay_interval] parameter.
   Scores are refreshed and decayed every [score_cleanup_ticks] heartbeats (see [limits]).

   TODO: https://gitlab.com/tezos/tezos/-/issues/5545
   We did not implement P6, aka IP colocation factor *)
type ('topic, 'span) score_limits = {
  topics : ('topic, 'span) topic_score_limits;
      (** Per-topic score parameters. *)
  topic_score_cap : float option;
      (** An optional cap on the total positive contribution of topics to the score of the peer.
          If not equal to [None], must be non-negative. *)
  behaviour_penalty_weight : float;
      (** P7: The weight of the score associated to the behaviour penalty. This
          parameter must be negative. *)
  behaviour_penalty_threshold : float;
      (** P7: The threshold on the behaviour penalty
          counter above which we start penalizing the peer. *)
  behaviour_penalty_decay : float;
      (** P7: The score is multiplied by a factor of [behaviour_penalty_decay] every [score_cleanup_ticks] heartbeat.
          This parameter must be in the unit interval. *)
  app_specific_weight : float;  (** P5: Application-specific peer scoring *)
  decay_zero : float;
      (** The minimum value under which a score is considered to be equal to 0 after
          applying decay. This parameter must be non-negative. *)
}

type ('topic, 'peer, 'message_id, 'span) limits = {
  max_recv_ihave_per_heartbeat : int;
      (** The maximum number of IHave control messages we can receive from a
          peer between two heartbeats. It is called [MaxIHaveMessages] in the Go
          implementation. *)
  max_sent_iwant_per_heartbeat : int;
      (** The maximum number of IWant control message ids we can send to a peer
          between two heartbeats. It is also the maximum number of message ids
          to include in an IHave message. It is called [MaxIHaveLength] in the
          Go implementation. *)
  max_gossip_retransmission : int;
      (** The maximum number of times the local peer allows a remote peer to
          request the same message id through IWant gossip before the local peer
          starts ignoring them. This is designed to prevent peers from spamming
          with requests. *)
  degree_optimal : int;
      (** The optimal number of full connections per topic. For
          example, if it is 6, each peer will want to have about six peers in
          their mesh for each topic they're subscribed to. It should be set
          somewhere between {!degree_low} and {!degree_high}. *)
  publish_threshold : float;
      (** The threshold value (as a score) from which we can publish a
          message to our peers. *)
  gossip_threshold : float;
      (** The threshold value (as a score) for a peer to emit/accept gossip: if
          the remote peer score is below this threshold, the local peer won't
          emit or accept gossip from the remote peer.  *)
  do_px : bool;  (** The flag controls whether peer exchange (PX) is enabled. *)
  accept_px_threshold : float;
      (** The threshold value (as a score) from which we accept peer exchanges. *)
  peers_to_px : int;
      (** The number of peers to include in prune Peer eXchange. (This is called
          [PrunePeers] in the Go implementation.) *)
  unsubscribe_backoff : 'span;
      (** The duration that prevent reconnections after leaving a topic to our full connections. *)
  graft_flood_threshold : 'span;
      (** If a graft comes before [graft_flood_threshold] has elapsed since the last prune,
          then there is an extra score penalty applied to the peer through P7. *)
  prune_backoff : 'span;  (** The duration added when we prune a peer. *)
  retain_duration : 'span;
      (** The duration added to remove metadata about a disconnected peer. *)
  fanout_ttl : 'span;
      (** [fanout_ttl] controls how long we keep track of a fanout topic. If
          it's been [fanout_ttl] since we've published to a topic that we're not
          subscribed to, then we don't track that topic anymore, that is, we
          delete it from the fanout map. *)
  heartbeat_interval : 'span;  (** The time between heartbeats. *)
  backoff_cleanup_ticks : int;
      (** The number of heartbeat ticks setting the frequency at which the
          backoffs are checked and potentially cleared. *)
  score_cleanup_ticks : int;
      (** [score_cleanup_ticks] is the number of heartbeat ticks setting the
          frequency at which the scores are refreshed and potentially cleared. *)
  degree_low : int;
      (** The lower bound on the number of peers we keep in a
          topic mesh. If we have fewer than [degree_low] peers, the heartbeat will attempt
          to graft some more into the mesh at the next heartbeat. *)
  degree_high : int;
      (** The upper bound on the number of peers we keep in a topic mesh. If
          there are more than [degree_high] peers, the heartbeat will select
          some to prune from the mesh at the next heartbeat. *)
  degree_score : int;
      (** [degree_score] affects how peers are selected when pruning a mesh due
          to over subscription. At least [degree_score] of the retained peers
          will be high-scoring, while the remainder are chosen randomly. *)
  (* TODO: https://gitlab.com/tezos/tezos/-/issues/5052
     [degree_score] must not exceed [degree_optimal - degree_out]. *)
  degree_out : int;
      (** The number of outbound connections to maintain in a topic mesh.  When
          the mesh is pruned due to over subscription, we make sure that we have
          outbound connections to at least [degree_out] of the survivor
          peers. This prevents Sybil attackers from overwhelming our mesh with
          incoming connections. [degree_out] must be set below {!degree_low},
          and must not exceed [degree_optimal / 2]. *)
  degree_lazy : int;
      (** [degree_lazy] affects how many peers the local peer will emit gossip
          to at each heartbeat. The local peer will send gossip to at least
          [degree_lazy] peers outside its mesh or fanout. The actual number may
          be more, depending on [gossip_factor] and how many peers the local
          peer is connected to. *)
  gossip_factor : float;
      (** [gossip_factor] affects how many peers the local peer will emit gossip
          to at each heartbeat. The local peer will send gossip to
          [gossip_factor] * (total number of non-mesh/non-fanout peers), or
          [degree_lazy] peers, whichever is greater. *)
  history_length : int;
      (** The size of the message cache used for gossip. The message cache will
          remember messages for [history_length] heartbeats. *)
  history_gossip_length : int;
      (** [history_gossip_length] controls how many cached message ids the local
          peer will advertise in IHave gossip messages. When asked for its seen
          message ids, the local peer will return only those from the most
          recent [history_gossip_length] heartbeats. The slack between
          [history_gossip_length] and [history_length] allows the local peer to
          avoid advertising messages that will be expired by the time they're
          requested. [history_gossip_length] must be less than or equal to
          [history_length]. *)
  opportunistic_graft_ticks : int64;
      (** The number of heartbeat ticks setting the frequency at which to
          attempt to improve the mesh with opportunistic grafting. Every
          [opportunistic_graft_ticks], if the median score of the mesh peers
          falls below the {!opportunistic_graft_threshold}, then the local peer
          will select some high-scoring mesh peers to graft.  *)
  opportunistic_graft_peers : int;
      (** The number of peers to opportunistically graft. *)
  opportunistic_graft_threshold : float;
      (** The median mesh score threshold before triggering opportunistic
          grafting; this should have a small positive value. *)
  score_limits : ('topic, 'span) score_limits;
      (** score-specific parameters. *)
  seen_history_length : int;
      (** [seen_history_length] controls the size of the message cache used for
          recording seen messages. The seen messages cache will remember messages for
          [seen_history_length] heartbeats. *)
}

type ('peer, 'message_id) parameters = {
  peer_filter :
    'peer -> [`IHave of 'message_id | `IWant of 'message_id | `Graft] -> bool;
}

(** The [SCORE] module type describes primitives used to update the scores
    associated to each peer. Score computation is described in more details in
    {{: https://github.com/libp2p/specs/blob/master/pubsub/gossipsub/gossipsub-v1.1.md#peer-scoring }the specification}.

    The score associated to a peer is a weighted sum of various counters, some
    of which are indexed by topics. Each captures a certain aspect of what a
    well-behaving peer should do. A short description follows, keeping the
    naming P1, P2, ... used in the specification.

    - P1: For each topic, we measure how long a peer has spent in the mesh for that topic. The longest, the better.
    - P2: For each topic, we measure how many times the peer was the first among all peers to send us a message on that topic.
    - P3: For each topic, we measure the number of message deliveries. If the number is below a threshold,
      we penalize the associated peer.
    - P3b: Trigger P3 computation when the peer gets pruned or removed, so as to not forget yet unaccounted for bad message
      delivery counts.
    - P4: For each topic, penalize peers sending invalid messages on that topic.
    - P5: The applicative layer can assign an arbitrary application-specific score to a peer.
    - P7: When a peer is pruned from the mesh for a topic, we install a backoff that
      prevents that peer from regrafting too soon. If the peer does not respect this backoff,
      it is penalized.
*)
module type SCORE = sig
  (** The type of peer scoring statistics. *)
  type t

  type value

  type span

  type time

  type topic

  (** [newly_connected params] creates a fresh statistics record. *)
  val newly_connected : (topic, span) score_limits -> t

  (** [value ps] evaluates the score of [ps]. *)
  val value : t -> value

  (** [penalty ps penalty] increments the behavioural penalty of [ps]. *)
  val penalty : t -> int -> t

  (** [set_connected ps] marks [ps] as being associated to a connected peer. *)
  val set_connected : t -> t

  (** [remove_peer ps ~retain_duration] will either return [None] if
      the peer statistics can be cleared, or [Some ps'] with [ps']
      some statistics to be retained for at least [retain_duration]. *)
  val remove_peer : t -> retain_duration:span -> t option

  (** [expires ps] returns [None] if the score statistics has no expiration
      time or [Some t] if it expires at time [t]. *)
  val expires : t -> time option

  (** [graft ps topic] allows to measure the time spent by the peer in the mesh.
      It is to be called upon grafting a peer to [topic]. *)
  val graft : t -> topic -> t

  (** [prune ps topic] allows to measure the time spent by the peer in the mesh.
      It is to be called upon pruning a peer from [topic]. *)
  val prune : t -> topic -> t

  (** [first_message_delivered ps topic] increments the counter related to
      first message deliveries and mesh message deliveries on [topic]
      by the associated peer. *)
  val first_message_delivered : t -> topic -> t

  (** [duplicate_message_delivered ps topic validated] increments the counter related to
      near-first mesh message deliveries on [topic] by the associated peer. [validated]
      is the time at which the message was seen by the automaton for the first time. *)
  val duplicate_message_delivered : t -> topic -> time -> t

  (** [invalid_message_delivered ps topic] increments the counter related to
      invalid messages sent by the associated peer. *)
  val invalid_message_delivered : t -> topic -> t

  (** [set_application_score ps score] sets the application-specific score. This score can
      be positive or negative.  *)
  val set_application_score : t -> float -> t

  (** [refresh ps] returns [Some ps'] with [ps'] a refreshed score record or [None]
      if the score expired. Refreshing a [ps] allows to update time-dependent spects
      of the scoring statistics. *)
  val refresh : t -> t option

  (** The zero score value, corresponding to a neutral score. *)
  val zero : value

  (** Convert a float into a score value. *)
  val of_float : float -> value

  include Compare.S with type t := value

  include PRINTABLE with type t := t

  val pp_value : Format.formatter -> value -> unit

  module Introspection : sig
    (** Convert a score value into a float.  *)
    val to_float : value -> float
  end

  module Internal_for_tests : sig
    val get_topic_params :
      ('topic, 'span) score_limits -> 'topic -> 'span per_topic_score_limits

    (** [is_active topic t] returns [true] if the peer's score for [topic] is marked as active,
        and [false] otherwise. *)
    val is_active : topic -> t -> bool
  end
end

module type MESSAGE_CACHE = sig
  module Peer : ITERABLE

  module Topic : ITERABLE

  module Message_id : ITERABLE

  module Message : PRINTABLE

  module Time : TIME

  (** A sliding window cache that stores published messages and their first seen
    time. The module also keeps track of the number of accesses to a message by
    a peer, thus indirectly tracking the number of IWant requests a peer makes
    for the same message between two heartbeats.

    The module assumes that no two different messages have the same message
    id. However, the cache stores duplicates; for instance, if [add_message id
    msg topic] is called exactly twice, then [msg] will appear twice in
    [get_message_ids_to_gossip]'s result (assuming not more than [gossip_slots]
    shifts have been executed in the meanwhile). *)
  type t

  (** [create ~history_slots ~gossip_slots ~seen_message_slots] creates two
      sliding window caches, one with length [history_slots] for storing message contents
      and another with length [seen_message] for storing seen messages and their first
      seen times.

      When queried for messages to advertise, the cache only returns messages in
      the last [gossip_slots]. The [gossip_slots] must be smaller or equal to
      [history_slots].

      The slack between [gossip_slots] and [history_slots] accounts for the
      reaction time between when a message is advertised via IHave gossip, and
      when the peer pulls it via an IWant command. To see this, if say
      [gossip_slot = history_slots] then the messages inserted in cache
      [history_slots] heartbeat ticks ago and advertised now will not be
      available after the next tick, because they are removed from the
      cache. This means IWant requests from the remote peer for such messages
      would be unfulfilled and potentially penalizing.

      @raise Assert_failure when [gossip_slots <= 0 || gossip_slots > history_slots]

      TODO: https://gitlab.com/tezos/tezos/-/issues/5129
      Error handling. *)
  val create :
    history_slots:int -> gossip_slots:int -> seen_message_slots:int -> t

  (** Add message to the most recent cache slot. If the message already exists
      in the cache, the message is not overridden, instead a duplicate message
      id is stored (the message itself is only stored once). *)
  val add_message : Message_id.t -> Message.t -> Topic.t -> t -> t

  (** Get the message associated to the given message id, increase the access
      counter for the peer requesting the message, and also returns the updated
      counter. *)
  val get_message_for_peer :
    Peer.t -> Message_id.t -> t -> (t * Message.t * int) option

  (** Get the message ids for the given topic in the last [gossip_slots] slots
      of the cache. If there were duplicates added in the cache, then there will
      be duplicates in the output. There is no guarantee about the order of
      messages in the output. *)
  val get_message_ids_to_gossip : Topic.t -> t -> Message_id.t list

  (** [get_first_seen_time message_id t] returns the time the message with [message_id]
      was first seen. Returns [None] if the message was not seen during the period
      covered by the sliding window. *)
  val get_first_seen_time : Message_id.t -> t -> Time.t option

  (** [seen_message message_id t] returns [true] if the message was seen during the
      period covered by the sliding window and returns [false] if otherwise. *)
  val seen_message : Message_id.t -> t -> bool

  (** Shift the sliding window by one slot (usually corresponding to one
      heartbeat tick). *)
  val shift : t -> t

  module Introspection : sig
    module Map : Map.S with type key = Int64.t

    val get_message_ids : t -> Message_id.t list Topic.Map.t Map.t
  end

  module Internal_for_tests : sig
    val get_access_counters : t -> (Message_id.t * int Peer.Map.t) Seq.t
  end
end

module type AUTOMATON = sig
  (** Module for peer *)
  module Peer : ITERABLE

  (** Module for topic *)
  module Topic : ITERABLE

  (** Module for message_id *)
  module Message_id : sig
    include ITERABLE

    (** Computes the topic of the given message id. *)
    val get_topic : t -> Topic.t
  end

  (** Module for message *)
  module Message : PRINTABLE

  (** Module for time *)
  module Time : PRINTABLE

  (** Module for time duration *)
  module Span : SPAN

  (** Module for peers scores *)
  module Score : SCORE with type time = Time.t and type topic = Topic.t

  type message = Message.t

  type span = Span.t

  (** The state managed by the gossipsub automaton. The state is
      purely functional. *)
  type state

  (** Limits of the gossipsub protocol. *)
  type limits := (Topic.t, Peer.t, Message_id.t, span) limits

  (** Parameters of the gossipsub protocol. *)
  type parameters := (Peer.t, Message_id.t) parameters

  (** The types of payloads for inputs to the gossipsub automaton. *)

  type add_peer = {direct : bool; outbound : bool; peer : Peer.t}

  type remove_peer = {peer : Peer.t}

  type ihave = {peer : Peer.t; topic : Topic.t; message_ids : Message_id.t list}

  type iwant = {peer : Peer.t; message_ids : Message_id.t list}

  type graft = {peer : Peer.t; topic : Topic.t}

  type prune = {
    peer : Peer.t;
    topic : Topic.t;
    px : Peer.t Seq.t;
    backoff : span;
  }

  type publish_message = {
    topic : Topic.t;
    message_id : Message_id.t;
    message : message;
  }

  type receive_message = {
    sender : Peer.t;
    topic : Topic.t;
    message_id : Message_id.t;
    message : message;
  }

  type join = {topic : Topic.t}

  type leave = {topic : Topic.t}

  type subscribe = {topic : Topic.t; peer : Peer.t}

  type unsubscribe = {topic : Topic.t; peer : Peer.t}

  type set_application_score = {peer : Peer.t; score : float}

  (** Output produced by one of the actions below. *)
  type _ output =
    | Ihave_from_peer_with_low_score : {
        score : Score.t;
        threshold : float;
      }
        -> [`IHave] output
        (** The peer who sent an IHave message has a [score] below [threshold]. *)
    | Too_many_recv_ihave_messages : {count : int; max : int} -> [`IHave] output
        (** The peer sent us more than [max] IHave messages within two
            successive heartbeat calls. *)
    | Too_many_sent_iwant_messages : {count : int; max : int} -> [`IHave] output
        (** We sent more than [max] IWant messages to this peer within two
            successive heartbeat calls. *)
    | Message_topic_not_tracked : [`IHave] output
        (** We received an IHave message for a topic we don't track. *)
    | Message_requested_message_ids : Message_id.t list -> [`IHave] output
        (** The messages ids we want to request from the peer which sent us an
            IHave message. The implementation honors the
            [max_sent_iwant_per_heartbeat] limit. *)
    | Invalid_message_id : [`IHave] output
        (** A message id received via IHave message is invalid. *)
    | Iwant_from_peer_with_low_score : {
        score : Score.t;
        threshold : float;
      }
        -> [`IWant] output
        (** The peer who sent an IWant message has a [score] below [threshold]. *)
    | On_iwant_messages_to_route : {
        routed_message_ids :
          [`Ignored | `Not_found | `Too_many_requests | `Message of message]
          Message_id.Map.t;
      }
        -> [`IWant] output
        (** As an answer for an [`IWant] message, the automaton returns a map
            associating to each requested message_id either [`Ignored] if the
            peer is filtered out by [peer_filter], [`Not_found] if the message
            is not found, or [Message m] if [m] is the message with the given
            id. *)
    | Peer_filtered : [`Graft] output
        (** The peer we attempt to graft has not been selected by
            [peer_filter]. *)
    | Unsubscribed_topic : [`Graft] output
        (** We didn't join the topic for which we are attempting to graft a
            peer. *)
    | Peer_already_in_mesh : [`Graft] output
        (** Attempting to graft a peer which has already been grafted. *)
    | Grafting_direct_peer : [`Graft] output
        (** Attempting to graft a direct peer. *)
    | Unexpected_grafting_peer : [`Graft] output
        (** The peer we attempt to graft is not known. *)
    | Grafting_peer_with_negative_score : [`Graft] output
        (** Attempting to graft a peer with a negative score. *)
    | Grafting_successfully : [`Graft] output
        (** Grafting the given peer for the provided topic succeeded. *)
    | Peer_backed_off : [`Graft] output
        (** We cannot graft the given peer because it is backed off. *)
    | Mesh_full : [`Graft] output
        (** Grafting a peer for a topic whose mesh has already sufficiently many
            peers. *)
    | Prune_topic_not_tracked : [`Prune] output
        (** Attempting to prune a peer for a non-tracked topic. *)
    | Peer_not_in_mesh : [`Prune] output
        (** Attempting to prune a peer which is not in the mesh. *)
    | Ignore_PX_score_too_low : Score.t -> [`Prune] output
        (** The given peer has been pruned for the given topic, but no
            alternative peers are returned because the peer's score is too low.
            The score of the peer is included in the return value. *)
    | No_PX : [`Prune] output
        (** The given peer has been pruned for the given topic. No
            alternatives peers was provided in {!prune}. *)
    | PX : Peer.Set.t -> [`Prune] output
        (** The given peer has been pruned for the given topic. The given set of
            peers alternatives in {!prune} for that topic is returned. *)
    | Publish_message : {to_publish : Peer.Set.t} -> [`Publish_message] output
        (** [to_publish] contains:
            - Direct peers for the message's topic;
            - The peers in the topic's mesh, if the peer is subscribed to the
              topic. Otherwise, the peers in the topic's fanout. *)
    | Already_published : [`Publish_message] output
        (** Attempting to publish a message that has already been published. *)
    | Route_message : {to_route : Peer.Set.t} -> [`Receive_message] output
        (** [to_route] contains:
            - Direct peers for the message's topic;
            - The peers in the topic's mesh minus the original sender of the message. *)
    | Already_received : [`Receive_message] output
        (** Received a message that has already been recevied before. *)
    | Not_subscribed : [`Receive_message] output
        (** Received a message from a remote peer for a topic we are not
            subscribed to (called "unknown topic" in the Go implementation). *)
    | Invalid_message : [`Receive_message] output
    | Unknown_validity : [`Receive_message] output
        (** Attempting to publish a message that is invalid. *)
    | Already_joined : [`Join] output
        (** Attempting to join a topic we already joined. *)
    | Joining_topic : {to_graft : Peer.Set.t} -> [`Join] output
        (** When successfully joining a topic, the set of grafted peers for that
            topic is returned. *)
    | Not_joined : [`Leave] output
        (** Attempting to leave a topic which we didn't join or had already
            left. *)
    | Leaving_topic : {
        to_prune : Peer.Set.t;
        noPX_peers : Peer.Set.t;
      }
        -> [`Leave] output
        (** When successfully leaving a topic, the set of pruned peers for that
            topic is returned alongside a subset of those peers for which no
            alternative PX will be proposed. *)
    | Heartbeat : {
        to_graft : Topic.Set.t Peer.Map.t;
            (** The set of topics per peer that have been grafted. *)
        to_prune : Topic.Set.t Peer.Map.t;
            (** The set of topics per peer that have been pruned. *)
        noPX_peers : Peer.Set.t;
            (** Set of peers for which peer exchange (PX) will not be
                proposed. *)
      }
        -> [`Heartbeat] output
    | Peer_added : [`Add_peer] output
        (** The output returned when successfully adding a peer. *)
    | Peer_already_known : [`Add_peer] output
        (** The output returned when attempting to add a peer which is already
            known. *)
    | Removing_peer : [`Remove_peer] output
        (** The output returned when successfully removing a peer. *)
    | Subscribed : [`Subscribe] output
        (** The output returned once we successfully processed a subscribe
            request sent from a peer. *)
    | Subscribe_to_unknown_peer : [`Subscribe] output
        (** The output returned when we receive a subscribe message from a peer
            we don't know.*)
    | Unsubscribed : [`Unsubscribe] output
        (** The output returned once we successfully processed an unsubscribe
            request sent from a peer. *)
    | Unsubscribe_from_unknown_peer : [`Unsubscribe] output
        (** The output returned when we receive an unsubscribe message from a
            peer we don't know.*)
    | Set_application_score : [`Set_application_score] output
        (** The output returned when we set the application score of a peer *)

  (** A type alias for the state monad. *)
  type 'a monad := state -> state * 'a output

  (** Initialise a state. *)
  val make : Random.State.t -> limits -> parameters -> state

  (** [add_peer { direct; outbound; peer }] is called to notify a new
      connection. If [direct] is [true], the gossipsub always forwards messages
      to those peers. [outbound] is [true] if it is an outbound connection, that
      is, a connection initiated by the local (not the remote) peer. Note
      however that the notion of "outbound" connections can be refined, relaxed
      or redefined by the application layer to fit its own needs. *)
  val add_peer : add_peer -> [`Add_peer] monad

  (** [remove_peer { peer }] notifies gossipsub that we are disconnected
      from a peer. Do note that the [state] still maintain information
      for this connection for [retain_duration] seconds. *)
  val remove_peer : remove_peer -> [`Remove_peer] monad

  (** [handle_subscribe {topic; peer}] handles a request from a remote [peer]
      informing us that it is subscribed to [topic]. *)
  val handle_subscribe : subscribe -> [`Subscribe] monad

  (** [handle_unsubscribe {topic; peer}] handles a request from a remote [peer]
      informing us that it unsubscribed from [topic]. *)
  val handle_unsubscribe : unsubscribe -> [`Unsubscribe] monad

  (** [handle_ihave { peer; topic; message_ids }] handles the gossip message
      [IHave] emitted by [peer] for [topic] with the [message_ids].  *)
  val handle_ihave : ihave -> [`IHave] monad

  (** [handle_iwant { peer; message_ids }] handles the gossip message
      [IWant] emitted by [peer] for [topic] with the [message_ids]. *)
  val handle_iwant : iwant -> [`IWant] monad

  (** [handle_graft { peer; topic }] handles the gossip message [Graft]
      emitted by [peer] for [topic]. This action allows to graft a
      connection to a full connection allowing the transmission of
      full messages for the given topic. *)
  val handle_graft : graft -> [`Graft] monad

  (** [handle_prune { peer; topic; px; backoff }] handles the gossip
      message [Prune] emitted by [peer] for [topic]. This action
      allows to prune a full connection. In that case, the remote peer
      can send a list of peers to connect to as well as a backoff
      time, which is a duration for which we cannot [Graft] this peer
      on this topic. *)
  val handle_prune : prune -> [`Prune] monad

  (** [handle_receive_message { sender; topic; message_id; message }] handles
      a message received from [sender] on the gossip network. The function returns
      a set of peers to which the (full) message will be directly forwarded.  *)
  val handle_receive_message : receive_message -> [`Receive_message] monad

  (** [publish { topic; message_id; message }] allows to publish a message
      on the gossip network from the local node. The function returns a set of peers
      to which the (full) message will be directly forwarded.  *)
  val publish_message : publish_message -> [`Publish_message] monad

  (** [heartbeat] executes the heartbeat routine of the algorithm. *)
  val heartbeat : [`Heartbeat] monad

  (** [join { topic }] handles a join to a new topic. On success, the function
      returns the set of peers that have been grafted to form the mesh of the
      joined topic. *)
  val join : join -> [`Join] monad

  (** [leave { topic }] handles a leave from a topic. On success, the function
      returns the set of peers, forming the mesh, that have been pruned for that
      topic. *)
  val leave : leave -> [`Leave] monad

  (** [set_application_score {peer; score}] handles setting the application score
      of [peer]. If the peer is not known, this does nothing. *)
  val set_application_score :
    set_application_score -> [`Set_application_score] monad

  (* TODO: https://gitlab.com/tezos/tezos/-/issues/5352
     Handle the random state explicitly for [select_px_peers] and
     [select_gossip_messages]. *)

  (** Select random peers for Peer eXchange. Note that function is
      deterministic; however, it has side effects in that it updates the
      [state]'s random state. *)
  val select_px_peers :
    state ->
    peer_to_prune:Peer.t ->
    Topic.t ->
    noPX_peers:Peer.Set.t ->
    Peer.t list

  (** Select the gossip messages to be sent. These are IHave control messages
      referring to recently seen messages (that is, sent during the last
      [history_gossip_length] heartbeat ticks), to be sent to a random selection
      of peers. The message ids for a peer and a topic are also selected at
      random among the possible ones. At most [max_sent_iwant_per_heartbeat]
      message ids are sent.

      The local peer will send gossip to at most [gossip_factor] * (total number
      of non-mesh/non-fanout peers), or [degree_lazy] random peers, whichever is
      greater.

      Note that function is deterministic; however, it has side effects in that
      it updates the [state]'s random state.  *)
  val select_gossip_messages : state -> ihave list

  val pp_add_peer : Format.formatter -> add_peer -> unit

  val pp_remove_peer : Format.formatter -> remove_peer -> unit

  val pp_ihave : Format.formatter -> ihave -> unit

  val pp_iwant : Format.formatter -> iwant -> unit

  val pp_graft : Format.formatter -> graft -> unit

  val pp_prune : Format.formatter -> prune -> unit

  val pp_receive_message : Format.formatter -> receive_message -> unit

  val pp_publish_message : Format.formatter -> publish_message -> unit

  val pp_join : Format.formatter -> join -> unit

  val pp_leave : Format.formatter -> leave -> unit

  val pp_subscribe : Format.formatter -> subscribe -> unit

  val pp_unsubscribe : Format.formatter -> unsubscribe -> unit

  val pp_set_application_score :
    Format.formatter -> set_application_score -> unit

  val pp_output : Format.formatter -> 'a output -> unit

  module Introspection : sig
    type connection = {topics : Topic.Set.t; direct : bool; outbound : bool}

    type fanout_peers = {peers : Peer.Set.t; last_published_time : Time.t}

    module Connections : sig
      type t

      val empty : t

      val bindings : t -> (Peer.t * connection) list

      val find : Peer.t -> t -> connection option

      val mem : Peer.t -> t -> bool

      val add_peer :
        Peer.t ->
        direct:bool ->
        outbound:bool ->
        t ->
        [`added of t | `already_known]

      val subscribe :
        Peer.t -> Topic.t -> t -> [`unknown_peer | `subscribed of t]

      val unsubscribe :
        Peer.t -> Topic.t -> t -> [`unknown_peer | `unsubscribed of t]

      val remove : Peer.t -> t -> t

      val fold : (Peer.t -> connection -> 'b -> 'b) -> t -> 'b -> 'b

      val iter : (Peer.t -> connection -> unit) -> t -> unit

      val peers_in_topic : Topic.t -> t -> Peer.Set.t

      val peers_per_topic_map : t -> Peer.Set.t Topic.Map.t
    end

    module Message_cache :
      MESSAGE_CACHE
        with type Peer.t = Peer.t
         and type 'a Peer.Map.t = 'a Peer.Map.t
         and type Topic.t = Topic.t
         and type Message_id.t = Message_id.t
         and type Message.t = Message.t
         and type Time.t = Time.t

    type view = {
      limits : limits;
      parameters : parameters;
      connections : Connections.t;
      scores : Score.t Peer.Map.t;
      ihave_per_heartbeat : int Peer.Map.t;
      iwant_per_heartbeat : int Peer.Map.t;
      mesh : Peer.Set.t Topic.Map.t;
      fanout : fanout_peers Topic.Map.t;
      backoff : Time.t Peer.Map.t Topic.Map.t;
      message_cache : Message_cache.t;
      rng : Random.State.t;
      heartbeat_ticks : int64;
    }

    (** When selecting a set of connected peers, one can specify some criteria
        to filter the result. *)
    type connected_peers_filter =
      | Direct
      | Subscribed_to of Topic.t
      | Score_above of {threshold : Score.value}

    val view : state -> view

    (** [get_peers_in_topic_mesh topic state] returns the peers in the mesh of
        [topic]. *)
    val get_peers_in_topic_mesh : Topic.t -> view -> Peer.t list

    (** [get_connected_peers ?filters view] returns the list of connected peers
        filtered by the given criteria.  *)
    val get_connected_peers :
      ?filters:connected_peers_filter list -> view -> Peer.t list

    (** [get_our_topics state] returns the set of topics the local peer is
        subscribed to. *)
    val get_our_topics : view -> Topic.t list

    (** [get_subscribed_topics peer state] returns the set of topics
        that are subscribed by [peer] *)
    val get_subscribed_topics : Peer.t -> view -> Topic.t list

    (** [get_fanout_peers topic state] returns the fanout peers of [topic]. *)
    val get_fanout_peers : Topic.t -> view -> Peer.t list

    (** [get_peer_score peer view] returns the score of [peer]. *)
    val get_peer_score : Peer.t -> view -> Score.value

    (** [get_peer_ihave_per_heartbeat peer view] returns
        the number of IHaves received from [peer] since the last heartbeat. *)
    val get_peer_ihave_per_heartbeat : Peer.t -> view -> int

    (** [get_peer_iwant_per_heartbeat peer view] returns
        the number of IWants sent to [peer] since the last heartbeat. *)
    val get_peer_iwant_per_heartbeat : Peer.t -> view -> int

    (** [get_peer_backoff topic peer view] returns the backoff time of [peer] for [topic].
        Returns [None] if the peer is not backoffed for [topic]. *)
    val get_peer_backoff : Topic.t -> Peer.t -> view -> Time.t option

    val limits : view -> limits

    (** [has_joined topic view] returns true if and only if the automaton is
        currently tracking messages for [topic]. That is, the local peer has joined
        and hasn't left the [topic]. *)
    val has_joined : Topic.t -> view -> bool

    (** [in_mesh peer topic view] returns true if and only if [peer] is
        in the mesh of [topic]. *)
    val in_mesh : Topic.t -> Peer.t -> view -> bool

    (** [is_direct peer view] returns true if and only if [peer] is a direct peer. *)
    val is_direct : Peer.t -> view -> bool

    (** [is_outbound peer view] returns true if and only if
        [peer] has an outbound connection. *)
    val is_outbound : Peer.t -> view -> bool

    val pp_connection : connection Fmt.t

    val pp_connections : Connections.t Fmt.t

    val pp_scores : Score.value Peer.Map.t Fmt.t

    val pp_peer_map : 'a Fmt.t -> 'a Peer.Map.t Fmt.t

    val pp_message_id_map : 'a Fmt.t -> 'a Message_id.Map.t Fmt.t

    val pp_topic_map : 'a Fmt.t -> 'a Topic.Map.t Fmt.t

    val pp_peer_set : Peer.Set.t Fmt.t

    val pp_topic_set : Topic.Set.t Fmt.t
  end
end

module type WORKER_CONFIGURATION = sig
  (** The gossipsub automaton that will be used by the worker. *)
  module GS : AUTOMATON

  module Point : ITERABLE

  (** Abstraction of the IO monad used by the worker. *)
  module Monad : sig
    (** The monad type. *)
    type 'a t

    (** Equivalent to [bind m f] function, in infix notation. *)
    val ( let* ) : 'a t -> ('a -> 'b t) -> 'b t

    (** The monad's return function. *)
    val return : 'a -> 'a t

    (** [sleep span] will block for the amount of time specified by [span]. *)
    val sleep : GS.span -> unit t
  end

  (** A mutable (FIFO) stream of data. *)
  module Stream : sig
    (** The stream data structure. *)
    type 'a t

    (** Create a new empty stream. *)
    val empty : unit -> 'a t

    (** Push the given value into the stream. *)
    val push : 'a -> 'a t -> unit

    (** Pops the oldest value that has been pushed to the stream. In case the
        stream is empty, the function will wait until some value is pushed. *)
    val pop : 'a t -> 'a Monad.t

    (** Returns and removes all available elements of the stream l without
        blocking. *)
    val get_available : 'a t -> 'a list

    (** Returns the number of elements in the stream. *)
    val length : 'a t -> int
  end
end

(** The interface of a gossipsub worker. *)
module type WORKER = sig
  (** The state of a gossipsub worker. *)
  type t

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

  (** A message together with a header, that is, a topic and an id.
     This corresponds to what the spec calls a "full message". *)
  type message_with_header = {
    message : GS.Message.t;
    topic : GS.Topic.t;
    message_id : GS.Message_id.t;
  }

  (** The following type defines the different kinds of messages a peer could
      receive from or sent to the P2P layer. *)
  type p2p_message =
    | Graft of {topic : GS.Topic.t}
    | Prune of {topic : GS.Topic.t; px : GS.Peer.t Seq.t; backoff : GS.Span.t}
    | IHave of {topic : GS.Topic.t; message_ids : GS.Message_id.t list}
    | IWant of {message_ids : GS.Message_id.t list}
    | Subscribe of {topic : GS.Topic.t}
    | Unsubscribe of {topic : GS.Topic.t}
    | Message_with_header of message_with_header

  (** The different kinds of input events that could be received from the P2P
      layer. *)
  type p2p_input =
    | In_message of {from_peer : GS.Peer.t; p2p_message : p2p_message}
    | New_connection of {
        peer : GS.Peer.t;
        direct : bool;
        trusted : bool;
        bootstrap : bool;
      }
    | Disconnection of {peer : GS.Peer.t}

  (** The different kinds of input events that could be received from the
      application layer. *)
  type app_input =
    | Publish_message of message_with_header
    | Join of GS.Topic.t
    | Leave of GS.Topic.t

  (** A peer's origin is either another peer (i.e. advertised via PX), or none
      if it is trusted. *)
  type peer_origin = PX of GS.Peer.t | Trusted

  (** The different kinds of outputs that could be emitted by the worker for the
      P2P layer. *)
  type p2p_output =
    | Out_message of {to_peer : GS.Peer.t; p2p_message : p2p_message}
        (** Emit the given [p2p_message] to the remote peer [to_peer]. *)
    | Disconnect of {peer : GS.Peer.t}
        (** End the connection with the peer [peer]. *)
    | Kick of {peer : GS.Peer.t}
        (** Kick the peer [peer]: the peer is disconnected and blacklisted.*)
    | Connect of {peer : GS.Peer.t; 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 {point : Point.t}
        (** Version of connect where we provide a point directly. *)
    (* TODO: https://gitlab.com/tezos/tezos/-/issues/6741

       Unify the two Connect versions. Have the peers cache in the worker. *)
    | Forget of {peer : GS.Peer.t; 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 application layer will be advertised about full messages it's
      interested in. *)
  type app_output = message_with_header

  (** The different kinds of events the Gossipsub worker handles. *)
  type event = private
    | Heartbeat
    | P2P_input of p2p_input
    | App_input of app_input

  (** [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 make :
    ?events_logging:(event -> unit Monad.t) ->
    ?bootstrap_points:Point.t list ->
    Random.State.t ->
    (GS.Topic.t, GS.Peer.t, GS.Message_id.t, GS.span) limits ->
    (GS.Peer.t, GS.Message_id.t) parameters ->
    t

  (** [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 start : GS.Topic.t list -> t -> unit

  (** [shutdown state] allows stopping the worker whose [state] is given. *)
  val shutdown : t -> unit Monad.t

  (** [app_input state app_input] adds the given application input [app_input]
      to the worker's input stream. *)
  val app_input : t -> app_input -> unit

  (** [p2p_input state p2p_input] adds the given P2P input [p2p_input] to the
      worker's input stream. *)
  val p2p_input : t -> p2p_input -> unit

  (** [p2p_output_stream t] returns the output stream containing data for the
      P2P layer. *)
  val p2p_output_stream : t -> p2p_output Stream.t

  (** [app_output_stream t] returns the output stream containing data for the
      application layer. *)
  val app_output_stream : t -> app_output Stream.t

  (** [input_events_stream t] returns the input stream in which we push events
      to be processed by the worker. *)
  val input_events_stream : t -> event Stream.t

  (** [is_subscribed t topic] checks whether [topic] is in the mesh of [t]. *)
  val is_subscribed : t -> GS.Topic.t -> bool

  (** Pretty-printer for values of type {!p2p_output}. *)
  val pp_p2p_output : Format.formatter -> p2p_output -> unit

  (** Pretty-printer for values of type {!app_output}. *)
  val pp_app_output : Format.formatter -> app_output -> unit

  (** Introspection and stats facilities *)
  module Introspection : sig
    (** A record containing some stats about what happened in the Gossipsub
        worker.  *)
    type stats = private {
      mutable count_topics : int64;
          (** Counts the number of topics of the node. It's the diff between Join
            and Leave topics events. *)
      mutable count_connections : int64;
          (** Counts the number of connections of the node. It's the diff between
            New_connection and Disconnection events. *)
      mutable count_bootstrap_connections : int64;
          (** Counts the number of connections of the node to bootstrap
            peers. It's a refinement of [count_connections] for when the remote
            peer declares itself as a bootstrap peer. *)
      mutable count_sent_app_messages : int64;  (** Count sent app messages. *)
      mutable count_sent_grafts : int64;  (** Count sent grafts. *)
      mutable count_sent_prunes : int64;  (** Count sent prunes. *)
      mutable count_sent_ihaves : int64;  (** Count sent ihaves. *)
      mutable count_sent_iwants : int64;  (** Count sent iwants. *)
      mutable count_recv_valid_app_messages : int64;
          (** Count received app messages that are known to be valid. *)
      mutable count_recv_invalid_app_messages : int64;
          (** Count received app messages that are known to be invalid. *)
      mutable count_recv_unknown_validity_app_messages : int64;
          (** Count received app messages we won't validate. *)
      mutable count_recv_grafts : int64;
          (** Count successfully received & processed grafts. *)
      mutable count_recv_prunes : int64;
          (** Count successfully received & processed prunes. *)
      mutable count_recv_ihaves : int64;
          (** Count successfully received & processed ihaves. *)
      mutable count_recv_iwants : int64;
          (** Count successfully received & processed iwants. *)
    }

    val empty_stats : unit -> stats
  end

  val stats : t -> Introspection.stats

  val state : t -> GS.Introspection.view
end