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

Coverage Stats

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

0.0

/src/lib/transition_frontier_controller/transition_frontier_controller.ml

open Core_kernel
open Async_kernel
open Pipe_lib
open Mina_block

module type CONTEXT = sig
  val logger : Logger.t

  val precomputed_values : Precomputed_values.t

  val constraint_constants : Genesis_constants.Constraint_constants.t

  val consensus_constants : Consensus.Constants.t

  val proof_cache_db : Proof_cache_tag.cache_db

  val signature_kind : Mina_signature_kind.t
end

let run ~context:(module Context : CONTEXT) ~trust_system ~verifier ~network
    ~time_controller ~collected_transitions ~frontier ~get_completed_work
    ~network_transition_reader ~producer_transition_reader ~clear_reader
    ~cache_exceptions ?transaction_pool_proxy =
  let open Context in
  let valid_transition_pipe_capacity = 50 in
  let start_time = Time.now () in
  let f_drop_head name head valid_cb =
    let hashes =
      match head with
      | `Block b ->
          With_hash.hash @@ Validation.block_with_hash
          @@ Network_peer.Envelope.Incoming.data @@ Cache_lib.Cached.peek b
      | `Header h ->
          With_hash.hash @@ Validation.header_with_hash
          @@ Network_peer.Envelope.Incoming.data h
    in
    Mina_block.handle_dropped_transition hashes ?valid_cb ~pipe_name:name
      ~logger
  in
  let valid_transition_reader, valid_transition_writer =
    let name = "valid transitions" in
    Strict_pipe.create ~name
      (Buffered
         ( `Capacity valid_transition_pipe_capacity
         , `Overflow
             (Drop_head
                (fun (head, `Valid_cb vc) ->
                  Mina_metrics.(
                    Counter.inc_one
                      Pipe.Drop_on_overflow
                      .transition_frontier_valid_transitions) ;
                  f_drop_head name head vc ) ) ) )
  in
  let primary_transition_pipe_capacity =
    valid_transition_pipe_capacity + List.length collected_transitions
  in
  (* Ok to drop on overflow- catchup will be triggered if required*)
  let primary_transition_reader, primary_transition_writer =
    let name = "primary transitions" in
    Strict_pipe.create ~name
      (Buffered
         ( `Capacity primary_transition_pipe_capacity
         , `Overflow
             (Drop_head
                (fun (head, `Valid_cb vc) ->
                  Mina_metrics.(
                    Counter.inc_one
                      Pipe.Drop_on_overflow
                      .transition_frontier_primary_transitions) ;
                  f_drop_head name head vc ) ) ) )
  in
  let processed_transition_reader, processed_transition_writer =
    Strict_pipe.create ~name:"processed transitions"
      (Buffered (`Capacity 30, `Overflow Crash))
  in
  let catchup_job_reader, catchup_job_writer =
    Strict_pipe.create ~name:"catchup jobs"
      (Buffered (`Capacity 30, `Overflow Crash))
  in
  let catchup_breadcrumbs_reader, catchup_breadcrumbs_writer =
    Strict_pipe.create ~name:"catchup breadcrumbs"
      (Buffered (`Capacity 30, `Overflow Crash))
  in
  let unprocessed_transition_cache =
    Transition_handler.Unprocessed_transition_cache.create ~logger
      ~cache_exceptions
  in
  List.iter collected_transitions ~f:(fun (t, vc) ->
      let b_or_h =
        match Network_peer.Envelope.Incoming.data t with
        | `Block b ->
            (* since the cache was just built, it's safe to assume
             * registering these will not fail, so long as there
             * are no duplicates in the list *)
            `Block
              ( Transition_handler.Unprocessed_transition_cache.register_exn
                  unprocessed_transition_cache
              @@ Network_peer.Envelope.Incoming.map ~f:(const b) t )
        | `Header h ->
            `Header (Network_peer.Envelope.Incoming.map ~f:(const h) t)
      in
      Strict_pipe.Writer.write primary_transition_writer (b_or_h, `Valid_cb vc) ) ;
  let initial_state_hashes =
    List.map collected_transitions ~f:(fun (envelope, _) ->
        Network_peer.Envelope.Incoming.data envelope
        |> Bootstrap_controller.Transition_cache.header_with_hash
        |> Mina_base.State_hash.With_state_hashes.state_hash )
    |> Mina_base.State_hash.Set.of_list
  in
  let extensions = Transition_frontier.extensions frontier in
  don't_wait_for
  @@ Pipe_lib.Broadcast_pipe.Reader.iter_until
       (Transition_frontier.Extensions.get_view_pipe extensions New_breadcrumbs)
       ~f:(fun new_breadcrumbs ->
         let open Mina_base.State_hash in
         let new_state_hashes =
           List.map new_breadcrumbs ~f:Transition_frontier.Breadcrumb.state_hash
           |> Set.of_list
         in
         if Set.is_empty @@ Set.inter initial_state_hashes new_state_hashes then
           Deferred.return false
         else (
           Mina_metrics.(
             Gauge.set Catchup.initial_catchup_time
               Time.(Span.to_min @@ diff (now ()) start_time)) ;
           Deferred.return true ) ) ;
  Transition_handler.Validator.run
    ~context:(module Context)
    ~trust_system ~time_controller ~frontier
    ~transition_reader:network_transition_reader ~valid_transition_writer
    ~unprocessed_transition_cache ;
  Strict_pipe.Reader.iter_without_pushback valid_transition_reader
    ~f:(Strict_pipe.Writer.write primary_transition_writer)
  |> don't_wait_for ;
  let clean_up_catchup_scheduler = Ivar.create () in
  Transition_handler.Processor.run
    ~context:(module Context)
    ~time_controller ~trust_system ~verifier ~frontier ~get_completed_work
    ~primary_transition_reader ~producer_transition_reader
    ~clean_up_catchup_scheduler ~catchup_job_writer ~catchup_breadcrumbs_reader
    ~catchup_breadcrumbs_writer ~processed_transition_writer
    ?transaction_pool_proxy ;
  Ledger_catchup.run
    ~context:(module Context)
    ~trust_system ~verifier ~network ~frontier ~catchup_job_reader
    ~catchup_breadcrumbs_writer ~unprocessed_transition_cache ;
  upon (Strict_pipe.Reader.read clear_reader) (fun _ ->
      let open Strict_pipe.Writer in
      kill valid_transition_writer ;
      kill primary_transition_writer ;
      kill processed_transition_writer ;
      kill catchup_job_writer ;
      kill catchup_breadcrumbs_writer ;
      if Ivar.is_full clean_up_catchup_scheduler then
        [%log error] "Ivar.fill bug is here!" ;
      Ivar.fill clean_up_catchup_scheduler () ) ;
  processed_transition_reader

1	open Core_kernel
2	open Async_kernel
3	open Pipe_lib
4	open Mina_block
5
6	module type CONTEXT = sig
7	val logger : Logger.t
8
9	val precomputed_values : Precomputed_values.t
10
11	val constraint_constants : Genesis_constants.Constraint_constants.t
12
13	val consensus_constants : Consensus.Constants.t
14
15	val proof_cache_db : Proof_cache_tag.cache_db
16
17	val signature_kind : Mina_signature_kind.t
18	end
19
20	let run ~context:(module Context : CONTEXT) ~trust_system ~verifier ~network
21	~time_controller ~collected_transitions ~frontier ~get_completed_work
22	~network_transition_reader ~producer_transition_reader ~clear_reader
23	~cache_exceptions ?transaction_pool_proxy =
UNCOV 24	let open Context in	×
25	let valid_transition_pipe_capacity = 50 in
26	let start_time = Time.now () in
UNCOV 27	let f_drop_head name head valid_cb =	×
28	let hashes =	×
29	match head with
30	\| `Block b ->	×
31	With_hash.hash @@ Validation.block_with_hash	×
32	@@ Network_peer.Envelope.Incoming.data @@ Cache_lib.Cached.peek b	×
33	\| `Header h ->	×
34	With_hash.hash @@ Validation.header_with_hash	×
35	@@ Network_peer.Envelope.Incoming.data h	×
36	in
37	Mina_block.handle_dropped_transition hashes ?valid_cb ~pipe_name:name
38	~logger
39	in
40	let valid_transition_reader, valid_transition_writer =
41	let name = "valid transitions" in
UNCOV 42	Strict_pipe.create ~name	×
43	(Buffered
44	( `Capacity valid_transition_pipe_capacity
45	, `Overflow
46	(Drop_head
47	(fun (head, `Valid_cb vc) ->
48	Mina_metrics.(	×
49	Counter.inc_one	×
50	Pipe.Drop_on_overflow
51	.transition_frontier_valid_transitions) ;
52	f_drop_head name head vc ) ) ) )
53	in
54	let primary_transition_pipe_capacity =
UNCOV 55	valid_transition_pipe_capacity + List.length collected_transitions	×
56	in
57	(* Ok to drop on overflow- catchup will be triggered if required*)
58	let primary_transition_reader, primary_transition_writer =
59	let name = "primary transitions" in
UNCOV 60	Strict_pipe.create ~name	×
61	(Buffered
62	( `Capacity primary_transition_pipe_capacity
63	, `Overflow
64	(Drop_head
65	(fun (head, `Valid_cb vc) ->
66	Mina_metrics.(	×
67	Counter.inc_one	×
68	Pipe.Drop_on_overflow
69	.transition_frontier_primary_transitions) ;
70	f_drop_head name head vc ) ) ) )
71	in
72	let processed_transition_reader, processed_transition_writer =
73	Strict_pipe.create ~name:"processed transitions"
74	(Buffered (`Capacity 30, `Overflow Crash))
75	in
UNCOV 76	let catchup_job_reader, catchup_job_writer =	×
77	Strict_pipe.create ~name:"catchup jobs"
78	(Buffered (`Capacity 30, `Overflow Crash))
79	in
UNCOV 80	let catchup_breadcrumbs_reader, catchup_breadcrumbs_writer =	×
81	Strict_pipe.create ~name:"catchup breadcrumbs"
82	(Buffered (`Capacity 30, `Overflow Crash))
83	in
UNCOV 84	let unprocessed_transition_cache =	×
85	Transition_handler.Unprocessed_transition_cache.create ~logger
86	~cache_exceptions
87	in
88	List.iter collected_transitions ~f:(fun (t, vc) ->
UNCOV 89	let b_or_h =	×
90	match Network_peer.Envelope.Incoming.data t with
UNCOV 91	\| `Block b ->	×
92	(* since the cache was just built, it's safe to assume
93	* registering these will not fail, so long as there
94	* are no duplicates in the list *)
95	`Block
96	( Transition_handler.Unprocessed_transition_cache.register_exn
UNCOV 97	unprocessed_transition_cache	×
UNCOV 98	@@ Network_peer.Envelope.Incoming.map ~f:(const b) t )	×
99	\| `Header h ->	×
100	`Header (Network_peer.Envelope.Incoming.map ~f:(const h) t)	×
101	in
102	Strict_pipe.Writer.write primary_transition_writer (b_or_h, `Valid_cb vc) ) ;
UNCOV 103	let initial_state_hashes =	×
104	List.map collected_transitions ~f:(fun (envelope, _) ->
UNCOV 105	Network_peer.Envelope.Incoming.data envelope	×
UNCOV 106	\|> Bootstrap_controller.Transition_cache.header_with_hash	×
UNCOV 107	\|> Mina_base.State_hash.With_state_hashes.state_hash )	×
UNCOV 108	\|> Mina_base.State_hash.Set.of_list	×
109	in
UNCOV 110	let extensions = Transition_frontier.extensions frontier in	×
UNCOV 111	don't_wait_for	×
UNCOV 112	@@ Pipe_lib.Broadcast_pipe.Reader.iter_until	×
UNCOV 113	(Transition_frontier.Extensions.get_view_pipe extensions New_breadcrumbs)	×
114	~f:(fun new_breadcrumbs ->
UNCOV 115	let open Mina_base.State_hash in	×
116	let new_state_hashes =
117	List.map new_breadcrumbs ~f:Transition_frontier.Breadcrumb.state_hash
UNCOV 118	\|> Set.of_list	×
119	in
UNCOV 120	if Set.is_empty @@ Set.inter initial_state_hashes new_state_hashes then	×
UNCOV 121	Deferred.return false	×
UNCOV 122	else (	×
123	Mina_metrics.(
UNCOV 124	Gauge.set Catchup.initial_catchup_time	×
UNCOV 125	Time.(Span.to_min @@ diff (now ()) start_time)) ;	×
126	Deferred.return true ) ) ;
UNCOV 127	Transition_handler.Validator.run	×
128	~context:(module Context)
129	~trust_system ~time_controller ~frontier
130	~transition_reader:network_transition_reader ~valid_transition_writer
131	~unprocessed_transition_cache ;
132	Strict_pipe.Reader.iter_without_pushback valid_transition_reader
UNCOV 133	~f:(Strict_pipe.Writer.write primary_transition_writer)	×
UNCOV 134	\|> don't_wait_for ;	×
UNCOV 135	let clean_up_catchup_scheduler = Ivar.create () in	×
UNCOV 136	Transition_handler.Processor.run	×
137	~context:(module Context)
138	~time_controller ~trust_system ~verifier ~frontier ~get_completed_work
139	~primary_transition_reader ~producer_transition_reader
140	~clean_up_catchup_scheduler ~catchup_job_writer ~catchup_breadcrumbs_reader
141	~catchup_breadcrumbs_writer ~processed_transition_writer
142	?transaction_pool_proxy ;
143	Ledger_catchup.run
144	~context:(module Context)
145	~trust_system ~verifier ~network ~frontier ~catchup_job_reader
146	~catchup_breadcrumbs_writer ~unprocessed_transition_cache ;
UNCOV 147	upon (Strict_pipe.Reader.read clear_reader) (fun _ ->	×
148	let open Strict_pipe.Writer in	×
149	kill valid_transition_writer ;
150	kill primary_transition_writer ;	×
151	kill processed_transition_writer ;	×
152	kill catchup_job_writer ;	×
153	kill catchup_breadcrumbs_writer ;	×
154	if Ivar.is_full clean_up_catchup_scheduler then	×
155	[%log error] "Ivar.fill bug is here!" ;	×
156	Ivar.fill clean_up_catchup_scheduler () ) ;	×
UNCOV 157	processed_transition_reader	×

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