MinaProtocol / mina / 767

Committed 04 Nov 2025 01:59PM UTC coverage: 32.374% (-4.5%) from 36.902%

Build # 767

Build Type

push

buildkite

Committed by

web-flow

Commit Message

Merge pull request #18063 from MinaProtocol/lyh/compat-into-dev-nov4-2025

Merge compatible into develop Nov. 4th 2025

Run Details

87 of 228 new or added lines in 10 files covered. (38.16%)

3416 existing lines in 136 files now uncovered.

23591 of 72871 relevant lines covered (32.37%)

26590.67 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

2.15

/src/lib/network_pool/network_pool_base.ml

open Async_kernel
open Core_kernel
open Pipe_lib
open Network_peer

module Make (Transition_frontier : sig
  type t
end)
(Resource_pool : Intf.Resource_pool_intf
                   with type transition_frontier := Transition_frontier.t) :
  Intf.Network_pool_base_intf
    with type resource_pool := Resource_pool.t
     and type resource_pool_diff := Resource_pool.Diff.t
     and type resource_pool_diff_verified := Resource_pool.Diff.verified
     and type transition_frontier := Transition_frontier.t
     and type transition_frontier_diff := Resource_pool.transition_frontier_diff
     and type config := Resource_pool.Config.t
     and type rejected_diff := Resource_pool.Diff.rejected = struct
  let apply_and_broadcast_thread_label =
    "apply_and_broadcast_" ^ Resource_pool.label ^ "_diffs"

  let apply_no_broadcast_thread_label =
    "apply_no_broadcast_" ^ Resource_pool.label ^ "_diffs"

  let processing_diffs_thread_label =
    "processing_" ^ Resource_pool.label ^ "_diffs"

  let processing_transition_frontier_diffs_thread_label =
    "processing_" ^ Resource_pool.label ^ "_transition_frontier_diffs"

  let rebroadcast_loop_thread_label = Resource_pool.label ^ "_rebroadcast_loop"

  module Broadcast_callback = struct
    type resource_pool_diff = Resource_pool.Diff.t

    type rejected_diff = Resource_pool.Diff.rejected

    type t =
      | Local of
          (   ( [ `Broadcasted | `Not_broadcasted ]
              * Resource_pool.Diff.t
              * Resource_pool.Diff.rejected )
              Or_error.t
           -> unit )
      | External of Mina_net2.Validation_callback.t

    let is_expired = function
      | Local _ ->
          false
      | External cb ->
          Mina_net2.Validation_callback.is_expired cb

    open Mina_net2.Validation_callback

    let error err = function
      | Local f ->
          f (Error err)
      | External cb ->
          fire_if_not_already_fired cb `Reject

    let reject accepted rejected = function
      | Local f ->
          f (Ok (`Not_broadcasted, accepted, rejected))
      | External cb ->
          fire_if_not_already_fired cb `Reject

    let drop accepted rejected = function
      | Local f ->
          f (Ok (`Not_broadcasted, accepted, rejected))
      | External cb ->
          fire_if_not_already_fired cb `Ignore

    let forward broadcast_pipe accepted rejected = function
      | Local f ->
          f (Ok (`Broadcasted, accepted, rejected)) ;
          Linear_pipe.write broadcast_pipe
            { With_nonce.message = accepted; nonce = 0 }
          |> don't_wait_for
      | External cb ->
          fire_if_not_already_fired cb `Accept
  end

  module Remote_sink =
    Pool_sink.Remote_sink
      (struct
        include Resource_pool.Diff

        let label = Resource_pool.label

        type pool = Resource_pool.t
      end)
      (Broadcast_callback)

  module Local_sink =
    Pool_sink.Local_sink
      (struct
        include Resource_pool.Diff

        let label = Resource_pool.label

        type pool = Resource_pool.t
      end)
      (Broadcast_callback)

  type t =
    { resource_pool : Resource_pool.t
    ; logger : Logger.t
    ; write_broadcasts : Resource_pool.Diff.t With_nonce.t Linear_pipe.Writer.t
    ; read_broadcasts : Resource_pool.Diff.t With_nonce.t Linear_pipe.Reader.t
    ; constraint_constants : Genesis_constants.Constraint_constants.t
    ; block_window_duration : Time.Span.t
    }

  let resource_pool { resource_pool; _ } = resource_pool

  let broadcasts { read_broadcasts; _ } = read_broadcasts

  let create_rate_limiter () =
    Rate_limiter.create
      ~capacity:
        (Resource_pool.Diff.max_per_15_seconds, `Per (Time.Span.of_sec 15.0))

  let apply_and_broadcast ({ logger; _ } as t)
      (diff : Resource_pool.Diff.verified Envelope.Incoming.t) cb =
    let env = Envelope.Incoming.map ~f:Resource_pool.Diff.t_of_verified diff in
    let rebroadcast (diff', rejected) =
      let open Broadcast_callback in
      if Resource_pool.Diff.is_empty diff' then (
        [%log trace]
          "Refusing to rebroadcast $diff. Pool diff apply feedback: empty diff"
          ~metadata:
            [ ( "diff"
              , `String Resource_pool.Diff.(summary @@ t_of_verified diff.data)
              )
            ] ;
        drop diff' rejected cb )
      else (
        [%log debug] "Rebroadcasting diff %s" (Resource_pool.Diff.summary diff') ;
        forward t.write_broadcasts diff' rejected cb )
    in
    O1trace.sync_thread apply_and_broadcast_thread_label (fun () ->
        match Resource_pool.Diff.unsafe_apply t.resource_pool diff with
        | Ok (`Accept, accepted, rejected) ->
            Resource_pool.Diff.log_internal ~logger "accepted" env ;
            rebroadcast (accepted, rejected)
        | Ok (`Reject, accepted, rejected) ->
            Resource_pool.Diff.log_internal ~logger "rejected"
              ~reason:"not_applied" env ;
            Broadcast_callback.reject accepted rejected cb
        | Error (`Locally_generated res) ->
            Resource_pool.Diff.log_internal ~logger "rejected"
              ~reason:"locally_generated" env ;
            rebroadcast res
        | Error (`Other e) ->
            [%log' debug t.logger]
              "Refusing to rebroadcast. Pool diff apply feedback: $error"
              ~metadata:[ ("error", Error_json.error_to_yojson e) ] ;
            Resource_pool.Diff.log_internal ~logger "rejected" env ;
            Broadcast_callback.error e cb )

  let apply_no_broadcast ({ logger; _ } as t)
      (diff : Resource_pool.Diff.verified Envelope.Incoming.t) =
    let env = Envelope.Incoming.map ~f:Resource_pool.Diff.t_of_verified diff in
    O1trace.sync_thread apply_no_broadcast_thread_label (fun () ->
        match Resource_pool.Diff.unsafe_apply t.resource_pool diff with
        | Ok (`Accept, _accepted, _rejected) ->
            Resource_pool.Diff.log_internal ~logger "accepted" env
        | Ok (`Reject, _accepted, _rejected) ->
            Resource_pool.Diff.log_internal ~logger "rejected"
              ~reason:"not_applied" env
        | Error (`Locally_generated _res) ->
            Resource_pool.Diff.log_internal ~logger "rejected"
              ~reason:"locally_generated" env
        | Error (`Other e) ->
            [%log' debug t.logger] "Pool diff apply feedback: $error"
              ~metadata:[ ("error", Error_json.error_to_yojson e) ] ;
            Resource_pool.Diff.log_internal ~logger "rejected" env )

  let log_rate_limiter_occasionally t rl =
    let time = Time_ns.Span.of_min 1. in
    every time (fun () ->
        [%log' debug t.logger]
          ~metadata:[ ("rate_limiter", Rate_limiter.summary rl) ]
          !"%s $rate_limiter" Resource_pool.label )

  type wrapped_t =
    | Diff of
        (Resource_pool.Diff.verified Envelope.Incoming.t * Broadcast_callback.t)
    | Transition_frontier_extension of Resource_pool.transition_frontier_diff

  let of_resource_pool_and_diffs resource_pool ~logger ~constraint_constants
      ~tf_diffs ~log_gossip_heard ~on_remote_push ~block_window_duration =
    let read_broadcasts, write_broadcasts = Linear_pipe.create () in
    let network_pool =
      { resource_pool
      ; logger
      ; read_broadcasts
      ; write_broadcasts
      ; constraint_constants
      ; block_window_duration
      }
    in
    let remote_r, remote_w, remote_rl =
      Remote_sink.create ~log_gossip_heard ~on_push:on_remote_push
        ~wrap:(fun m -> Diff m)
        ~unwrap:(function
          | Diff m -> m | _ -> failwith "unexpected message type" )
        ~trace_label:Resource_pool.label ~logger resource_pool
    in
    let local_r, local_w, _ =
      Local_sink.create
        ~wrap:(fun m -> Diff m)
        ~unwrap:(function
          | Diff m -> m | _ -> failwith "unexpected message type" )
        ~trace_label:Resource_pool.label ~logger resource_pool
    in
    log_rate_limiter_occasionally network_pool remote_rl ;
    (*priority: Transition frontier diffs > local diffs > incoming diffs*)
    Deferred.don't_wait_for
      (O1trace.thread Resource_pool.label (fun () ->
           Strict_pipe.Reader.Merge.iter_sync
             [ Strict_pipe.Reader.map tf_diffs ~f:(fun diff ->
                   Transition_frontier_extension diff )
             ; remote_r
             ; local_r
             ]
             ~f:(fun diff_source ->
               match diff_source with
               | Diff ((verified_diff, cb) : Remote_sink.unwrapped_t) ->
                   O1trace.sync_thread processing_diffs_thread_label (fun () ->
                       apply_and_broadcast network_pool verified_diff cb )
               | Transition_frontier_extension diff ->
                   O1trace.sync_thread
                     processing_transition_frontier_diffs_thread_label
                     (fun () ->
                       Resource_pool.handle_transition_frontier_diff diff
                         resource_pool ) ) ) ) ;
    (network_pool, remote_w, local_w)

  (* Rebroadcast locally generated pool items every 10 minutes. Do so for 50
     minutes - at most 5 rebroadcasts - before giving up.

     The goal here is to be resilient to short term network failures and
     partitions. Note that with gossip we don't know anything about the state of
     other peers' pools (we know if something made it into a block, but that can
     take a long time and it's possible for things to be successfully received
     but never used in a block), so in a healthy network all repetition is spam.
     We need to balance reliability with efficiency. Exponential "backoff" would
     be better, but it'd complicate the interface between this module and the
     specific pool implementations.
  *)
  let rebroadcast_loop : t -> Logger.t -> unit Deferred.t =
   fun t logger ->
    let rebroadcast_interval = Time.Span.of_min 10. in
    let rebroadcast_window = Time.Span.scale rebroadcast_interval 5. in
    let has_timed_out time =
      if Time.(add time rebroadcast_window < now ()) then `Timed_out else `Ok
    in
    let rec go () =
      let rebroadcastable =
        Resource_pool.get_rebroadcastable t.resource_pool ~has_timed_out
      in
      if List.is_empty rebroadcastable then
        [%log trace] "Nothing to rebroadcast"
      else
        [%log debug]
          "Preparing to rebroadcast locally generated resource pool diffs \
           $diffs"
          ~metadata:
            [ ("count", `Int (List.length rebroadcastable))
            ; ( "diffs"
              , `List
                  (List.map
                     ~f:(fun d -> `String (Resource_pool.Diff.summary d))
                     rebroadcastable ) )
            ] ;
      let nonce = Time_ns.to_int_ns_since_epoch (Time_ns.now ()) in
      let%bind () =
        Deferred.List.iter rebroadcastable ~f:(fun message ->
            Linear_pipe.write t.write_broadcasts With_nonce.{ message; nonce } )
      in
      let%bind () = Async.after rebroadcast_interval in
      go ()
    in
    go ()

  let create ~config ~constraint_constants ~consensus_constants ~time_controller
      ~frontier_broadcast_pipe ~logger ~log_gossip_heard ~on_remote_push
      ~block_window_duration =
    (* Diffs from transition frontier extensions *)
    let tf_diff_reader, tf_diff_writer =
      Strict_pipe.(
        create ~name:"Network pool transition frontier diffs" Synchronous)
    in
    let t, remotes, locals =
      of_resource_pool_and_diffs
        (Resource_pool.create ~constraint_constants ~consensus_constants
           ~time_controller ~config ~logger ~frontier_broadcast_pipe
           ~tf_diff_writer )
        ~constraint_constants ~logger ~tf_diffs:tf_diff_reader ~log_gossip_heard
        ~on_remote_push ~block_window_duration
    in
    O1trace.background_thread rebroadcast_loop_thread_label (fun () ->
        rebroadcast_loop t logger ) ;
    (t, remotes, locals)
end

1	open Async_kernel	122✔
2	open Core_kernel
3	open Pipe_lib
4	open Network_peer
5
6	module Make (Transition_frontier : sig
7	type t
8	end)
9	(Resource_pool : Intf.Resource_pool_intf
10	with type transition_frontier := Transition_frontier.t) :
11	Intf.Network_pool_base_intf
12	with type resource_pool := Resource_pool.t
13	and type resource_pool_diff := Resource_pool.Diff.t
14	and type resource_pool_diff_verified := Resource_pool.Diff.verified
15	and type transition_frontier := Transition_frontier.t
16	and type transition_frontier_diff := Resource_pool.transition_frontier_diff
17	and type config := Resource_pool.Config.t
18	and type rejected_diff := Resource_pool.Diff.rejected = struct
19	let apply_and_broadcast_thread_label =
20	"apply_and_broadcast_" ^ Resource_pool.label ^ "_diffs"
21
22	let apply_no_broadcast_thread_label =
23	"apply_no_broadcast_" ^ Resource_pool.label ^ "_diffs"
24
25	let processing_diffs_thread_label =
26	"processing_" ^ Resource_pool.label ^ "_diffs"
27
28	let processing_transition_frontier_diffs_thread_label =
29	"processing_" ^ Resource_pool.label ^ "_transition_frontier_diffs"
30
31	let rebroadcast_loop_thread_label = Resource_pool.label ^ "_rebroadcast_loop"
32
33	module Broadcast_callback = struct
34	type resource_pool_diff = Resource_pool.Diff.t
35
36	type rejected_diff = Resource_pool.Diff.rejected
37
38	type t =
39	\| Local of
40	( ( [ `Broadcasted \| `Not_broadcasted ]
41	* Resource_pool.Diff.t
42	* Resource_pool.Diff.rejected )
43	Or_error.t
44	-> unit )
45	\| External of Mina_net2.Validation_callback.t
46
47	let is_expired = function
48	\| Local _ ->	×
49	false
50	\| External cb ->	×
51	Mina_net2.Validation_callback.is_expired cb
52
53	open Mina_net2.Validation_callback
54
55	let error err = function
56	\| Local f ->	×
57	f (Error err)
58	\| External cb ->	×
59	fire_if_not_already_fired cb `Reject
60
61	let reject accepted rejected = function
62	\| Local f ->	×
63	f (Ok (`Not_broadcasted, accepted, rejected))
64	\| External cb ->	×
65	fire_if_not_already_fired cb `Reject
66
67	let drop accepted rejected = function
68	\| Local f ->	×
69	f (Ok (`Not_broadcasted, accepted, rejected))
70	\| External cb ->	×
71	fire_if_not_already_fired cb `Ignore
72
73	let forward broadcast_pipe accepted rejected = function
74	\| Local f ->	×
75	f (Ok (`Broadcasted, accepted, rejected)) ;
76	Linear_pipe.write broadcast_pipe	×
77	{ With_nonce.message = accepted; nonce = 0 }
78	\|> don't_wait_for
79	\| External cb ->	×
80	fire_if_not_already_fired cb `Accept
81	end
82
83	module Remote_sink =
84	Pool_sink.Remote_sink
85	(struct
86	include Resource_pool.Diff
87
88	let label = Resource_pool.label
89
90	type pool = Resource_pool.t
91	end)
92	(Broadcast_callback)
93
94	module Local_sink =
95	Pool_sink.Local_sink
96	(struct
97	include Resource_pool.Diff
98
99	let label = Resource_pool.label
100
101	type pool = Resource_pool.t
102	end)
103	(Broadcast_callback)
104
105	type t =
106	{ resource_pool : Resource_pool.t
107	; logger : Logger.t
108	; write_broadcasts : Resource_pool.Diff.t With_nonce.t Linear_pipe.Writer.t
109	; read_broadcasts : Resource_pool.Diff.t With_nonce.t Linear_pipe.Reader.t
110	; constraint_constants : Genesis_constants.Constraint_constants.t
111	; block_window_duration : Time.Span.t
112	}
113
114	let resource_pool { resource_pool; _ } = resource_pool	×
115
116	let broadcasts { read_broadcasts; _ } = read_broadcasts	×
117
118	let create_rate_limiter () =
119	Rate_limiter.create	×
120	~capacity:
121	(Resource_pool.Diff.max_per_15_seconds, `Per (Time.Span.of_sec 15.0))	×
122
123	let apply_and_broadcast ({ logger; _ } as t)
124	(diff : Resource_pool.Diff.verified Envelope.Incoming.t) cb =
125	let env = Envelope.Incoming.map ~f:Resource_pool.Diff.t_of_verified diff in	×
126	let rebroadcast (diff', rejected) =	×
127	let open Broadcast_callback in	×
128	if Resource_pool.Diff.is_empty diff' then (	×
129	[%log trace]	×
130	"Refusing to rebroadcast $diff. Pool diff apply feedback: empty diff"
131	~metadata:
132	[ ( "diff"
133	, `String Resource_pool.Diff.(summary @@ t_of_verified diff.data)	×
134	)
135	] ;
136	drop diff' rejected cb )	×
137	else (	×
138	[%log debug] "Rebroadcasting diff %s" (Resource_pool.Diff.summary diff') ;	×
139	forward t.write_broadcasts diff' rejected cb )	×
140	in
141	O1trace.sync_thread apply_and_broadcast_thread_label (fun () ->
142	match Resource_pool.Diff.unsafe_apply t.resource_pool diff with	×
143	\| Ok (`Accept, accepted, rejected) ->	×
144	Resource_pool.Diff.log_internal ~logger "accepted" env ;
145	rebroadcast (accepted, rejected)	×
146	\| Ok (`Reject, accepted, rejected) ->	×
147	Resource_pool.Diff.log_internal ~logger "rejected"
148	~reason:"not_applied" env ;
149	Broadcast_callback.reject accepted rejected cb	×
150	\| Error (`Locally_generated res) ->	×
151	Resource_pool.Diff.log_internal ~logger "rejected"
152	~reason:"locally_generated" env ;
153	rebroadcast res	×
154	\| Error (`Other e) ->	×
155	[%log' debug t.logger]	×
156	"Refusing to rebroadcast. Pool diff apply feedback: $error"
157	~metadata:[ ("error", Error_json.error_to_yojson e) ] ;	×
158	Resource_pool.Diff.log_internal ~logger "rejected" env ;	×
159	Broadcast_callback.error e cb )	×
160
161	let apply_no_broadcast ({ logger; _ } as t)
162	(diff : Resource_pool.Diff.verified Envelope.Incoming.t) =
163	let env = Envelope.Incoming.map ~f:Resource_pool.Diff.t_of_verified diff in	×
164	O1trace.sync_thread apply_no_broadcast_thread_label (fun () ->	×
165	match Resource_pool.Diff.unsafe_apply t.resource_pool diff with	×
166	\| Ok (`Accept, _accepted, _rejected) ->	×
167	Resource_pool.Diff.log_internal ~logger "accepted" env
168	\| Ok (`Reject, _accepted, _rejected) ->	×
169	Resource_pool.Diff.log_internal ~logger "rejected"
170	~reason:"not_applied" env
171	\| Error (`Locally_generated _res) ->	×
172	Resource_pool.Diff.log_internal ~logger "rejected"
173	~reason:"locally_generated" env
174	\| Error (`Other e) ->	×
175	[%log' debug t.logger] "Pool diff apply feedback: $error"	×
176	~metadata:[ ("error", Error_json.error_to_yojson e) ] ;	×
177	Resource_pool.Diff.log_internal ~logger "rejected" env )	×
178
179	let log_rate_limiter_occasionally t rl =
UNCOV 180	let time = Time_ns.Span.of_min 1. in	×
UNCOV 181	every time (fun () ->	×
UNCOV 182	[%log' debug t.logger]	×
UNCOV 183	~metadata:[ ("rate_limiter", Rate_limiter.summary rl) ]	×
184	!"%s $rate_limiter" Resource_pool.label )
185
186	type wrapped_t =
187	\| Diff of
188	(Resource_pool.Diff.verified Envelope.Incoming.t * Broadcast_callback.t)
189	\| Transition_frontier_extension of Resource_pool.transition_frontier_diff
190
191	let of_resource_pool_and_diffs resource_pool ~logger ~constraint_constants
192	~tf_diffs ~log_gossip_heard ~on_remote_push ~block_window_duration =
UNCOV 193	let read_broadcasts, write_broadcasts = Linear_pipe.create () in	×
UNCOV 194	let network_pool =	×
195	{ resource_pool
196	; logger
197	; read_broadcasts
198	; write_broadcasts
199	; constraint_constants
200	; block_window_duration
201	}
202	in
203	let remote_r, remote_w, remote_rl =
204	Remote_sink.create ~log_gossip_heard ~on_push:on_remote_push
205	~wrap:(fun m -> Diff m)	×
206	~unwrap:(function
207	\| Diff m -> m \| _ -> failwith "unexpected message type" )	×
208	~trace_label:Resource_pool.label ~logger resource_pool
209	in
UNCOV 210	let local_r, local_w, _ =	×
211	Local_sink.create
212	~wrap:(fun m -> Diff m)	×
213	~unwrap:(function
214	\| Diff m -> m \| _ -> failwith "unexpected message type" )	×
215	~trace_label:Resource_pool.label ~logger resource_pool
216	in
UNCOV 217	log_rate_limiter_occasionally network_pool remote_rl ;	×
218	(priority: Transition frontier diffs > local diffs > incoming diffs)
UNCOV 219	Deferred.don't_wait_for	×
UNCOV 220	(O1trace.thread Resource_pool.label (fun () ->	×
UNCOV 221	Strict_pipe.Reader.Merge.iter_sync	×
UNCOV 222	[ Strict_pipe.Reader.map tf_diffs ~f:(fun diff ->	×
UNCOV 223	Transition_frontier_extension diff )	×
224	; remote_r
225	; local_r
226	]
227	~f:(fun diff_source ->
UNCOV 228	match diff_source with	×
229	\| Diff ((verified_diff, cb) : Remote_sink.unwrapped_t) ->	×
230	O1trace.sync_thread processing_diffs_thread_label (fun () ->
231	apply_and_broadcast network_pool verified_diff cb )	×
UNCOV 232	\| Transition_frontier_extension diff ->	×
233	O1trace.sync_thread
234	processing_transition_frontier_diffs_thread_label
235	(fun () ->
UNCOV 236	Resource_pool.handle_transition_frontier_diff diff	×
237	resource_pool ) ) ) ) ;
UNCOV 238	(network_pool, remote_w, local_w)	×
239
240	(* Rebroadcast locally generated pool items every 10 minutes. Do so for 50
241	minutes - at most 5 rebroadcasts - before giving up.
242
243	The goal here is to be resilient to short term network failures and
244	partitions. Note that with gossip we don't know anything about the state of
245	other peers' pools (we know if something made it into a block, but that can
246	take a long time and it's possible for things to be successfully received
247	but never used in a block), so in a healthy network all repetition is spam.
248	We need to balance reliability with efficiency. Exponential "backoff" would
249	be better, but it'd complicate the interface between this module and the
250	specific pool implementations.
251	*)
252	let rebroadcast_loop : t -> Logger.t -> unit Deferred.t =
253	fun t logger ->
UNCOV 254	let rebroadcast_interval = Time.Span.of_min 10. in	×
UNCOV 255	let rebroadcast_window = Time.Span.scale rebroadcast_interval 5. in	×
UNCOV 256	let has_timed_out time =	×
257	if Time.(add time rebroadcast_window < now ()) then `Timed_out else `Ok	×
258	in
259	let rec go () =
UNCOV 260	let rebroadcastable =	×
261	Resource_pool.get_rebroadcastable t.resource_pool ~has_timed_out
262	in
UNCOV 263	if List.is_empty rebroadcastable then	×
UNCOV 264	[%log trace] "Nothing to rebroadcast"	×
265	else
266	[%log debug]	×
267	"Preparing to rebroadcast locally generated resource pool diffs \
268	$diffs"
269	~metadata:
270	[ ("count", `Int (List.length rebroadcastable))	×
271	; ( "diffs"
272	, `List
273	(List.map	×
274	~f:(fun d -> `String (Resource_pool.Diff.summary d))	×
275	rebroadcastable ) )
276	] ;
UNCOV 277	let nonce = Time_ns.to_int_ns_since_epoch (Time_ns.now ()) in	×
278	let%bind () =
UNCOV 279	Deferred.List.iter rebroadcastable ~f:(fun message ->	×
280	Linear_pipe.write t.write_broadcasts With_nonce.{ message; nonce } )	×
281	in
UNCOV 282	let%bind () = Async.after rebroadcast_interval in	×
283	go ()	×
284	in
285	go ()
286
287	let create ~config ~constraint_constants ~consensus_constants ~time_controller
288	~frontier_broadcast_pipe ~logger ~log_gossip_heard ~on_remote_push
289	~block_window_duration =
290	(* Diffs from transition frontier extensions *)
UNCOV 291	let tf_diff_reader, tf_diff_writer =	×
292	Strict_pipe.(
UNCOV 293	create ~name:"Network pool transition frontier diffs" Synchronous)	×
294	in
295	let t, remotes, locals =
296	of_resource_pool_and_diffs
297	(Resource_pool.create ~constraint_constants ~consensus_constants
298	~time_controller ~config ~logger ~frontier_broadcast_pipe
299	~tf_diff_writer )
300	~constraint_constants ~logger ~tf_diffs:tf_diff_reader ~log_gossip_heard
301	~on_remote_push ~block_window_duration
302	in
UNCOV 303	O1trace.background_thread rebroadcast_loop_thread_label (fun () ->	×
UNCOV 304	rebroadcast_loop t logger ) ;	×
UNCOV 305	(t, remotes, locals)	×
306	end	122✔

MinaProtocol / mina / 767

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous