Source file gossipsub_intf.ml
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
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. *)
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. *)
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
(** 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". *)
(** 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}
(** 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. *)
| 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