MinaProtocol / mina / 431

Committed 29 Jul 2025 03:33PM UTC coverage: 32.152% (-1.1%) from 33.242%

Build # 431

Build Type

push

buildkite

Committed by

web-flow

Commit Message

Merge pull request #17590 from MinaProtocol/georgeee/merge-compatible-to-develop-29jul2025

Merge compatible to develop (29 Jul 2025)

Run Details

80 of 143 new or added lines in 20 files covered. (55.94%)

655 existing lines in 25 files now uncovered.

23110 of 71877 relevant lines covered (32.15%)

24858.75 hits per line

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

88.52

/src/test/archive/archive_node_tests/archive_node_tests.ml

open Async
open Core
open Mina_automation
open Mina_automation_runner
open Mina_automation_fixture.Archive

(**
 * Test the basic functionality of the mina archive with mocked deamon
 *)

(* asserts count of archive blocked (we are skipping genesis block) *)
let assert_archived_blocks ~archive_uri ~expected =
  let connection = Psql.Conn_str archive_uri in
  let%bind actual_blocks_count =
    Psql.run_command ~connection
      "SELECT COUNT(*) FROM blocks WHERE global_slot_since_genesis > 1"
  in
  let actual_blocks_count =
    match actual_blocks_count with
    | Ok count ->
        count |> Int.of_string
    | Error err ->
        failwith ("Failed to query blocks count: " ^ Error.to_string_hum err)
  in
  if Int.( <> ) actual_blocks_count expected then
    failwithf "Invalid number of archive blocks. Actual (%d) vs Expected (%d)"
      actual_blocks_count expected ()
  else Deferred.unit

(* Convert performance metrics to a JSON format suitable for output *)
(* The metrics are expected to be a list of tuples (operation, avg_time) *)
(* where operation is a string and avg_time is a float representing the average time in milliseconds *)
let perf_metrics_to_yojson metrics =
  let json_list =
    List.map metrics ~f:(fun (operation, avg_time) ->
        `Assoc
          [ ("operation", `String operation); ("avg_time_ms", `Float avg_time) ] )
  in
  `List json_list

(** Extract performance metrics from a log file and calculate average execution times.

  This function reads a log file line by line, parses each line as a JSON log entry,
  and extracts performance metrics identified by the "is_perf_metric" metadata field.
  For each performance metric entry, it extracts the "elapsed" time and "label" fields.

  @param log_file Path to the log file to process
  @return A deferred list of tuples containing (operation_label, average_time_in_ms)
  
  The function performs the following steps:
  1. Reads all lines from the specified log file
  2. Filters and parses lines containing performance metrics
  3. Groups metrics by operation label
  4. Calculates the average execution time for each operation
  
  @raises Failure if a log line cannot be parsed as valid JSON
  @raises exn if required metadata fields ("elapsed" or "label") are missing *)
let extract_perf_metrics log_file =
  let open Deferred.Let_syntax in
  let%bind lines = Reader.file_lines log_file in
  let perf_metrics =
    List.filter_map lines ~f:(fun line ->
        if String.is_empty line then None
        else
          match Logger.Message.of_yojson (Yojson.Safe.from_string line) with
          | Ok entry ->
              if String.Map.mem entry.metadata "is_perf_metric" then
                let time_in_ms =
                  String.Map.find entry.metadata "elapsed"
                  |> Option.value_exn
                       ~message:
                         ("Missing elapsed in log entry in log line: " ^ line)
                  |> Yojson.Safe.Util.to_float
                in
                let label =
                  String.Map.find entry.metadata "label"
                  |> Option.value_exn
                       ~message:
                         ("Missing label in log entry in log line: " ^ line)
                  |> Yojson.Safe.Util.to_string
                in
                Some (label, time_in_ms)
              else None
          | Error err ->
              failwithf "Invalid log line: %s. Error: %s" line err () )
  in
  (* Calculate the average time for each operation *)
  (* Group by operation and calculate the average time *)
  let averaged_metrics =
    String.Map.of_alist_multi perf_metrics
    |> Map.to_alist
    |> List.map ~f:(fun (operation, times) ->
           let avg_time =
             List.fold times ~init:0.0 ~f:( +. )
             /. Float.of_int (List.length times)
           in
           (operation, avg_time) )
  in
  Deferred.return averaged_metrics

module ArchivePrecomputedBlocksFromDaemon = struct
  type t = Mina_automation_fixture.Archive.after_bootstrap

  let test_case (test_data : t) =
    let open Deferred.Let_syntax in
    let daemon = Daemon.default in
    let archive_uri = test_data.archive.config.postgres_uri in
    let output = test_data.temp_dir in
    let%bind precomputed_blocks =
      Network_data.untar_precomputed_blocks test_data.network_data output
    in
    let precomputed_blocks =
      List.map precomputed_blocks ~f:(fun file -> output ^ "/" ^ file)
      |> List.filter ~f:(fun file -> String.is_suffix file ~suffix:".json")
    in
    Archive.Process.start_logging test_data.archive
      ~log_file:(output ^ "/archive.log") ;
    let%bind () =
      Daemon.archive_blocks_from_files daemon
        ~archive_address:test_data.archive.config.server_port
        ~format:Archive_blocks.Precomputed precomputed_blocks
    in

    let%bind () =
      assert_archived_blocks ~archive_uri
        ~expected:(List.length precomputed_blocks)
    in
    let connection = Psql.Conn_str archive_uri in
    let%bind latest_state_hash =
      Psql.run_command ~connection
        "SELECT state_hash FROM blocks ORDER BY id DESC LIMIT 1"
    in
    let latest_state_hash =
      match latest_state_hash with
      | Ok hash ->
          hash
      | Error err ->
          failwith
            ("Failed to query latest state hash: " ^ Error.to_string_hum err)
    in
    let output_ledger = output ^ "/output_ledger.json" in
    let replayer = Replayer.default in
    let%bind replayer_output =
      Replayer.run replayer ~archive_uri
        ~input_config:
          (Network_data.replayer_input_file_path test_data.network_data)
        ~target_state_hash:latest_state_hash ~interval_checkpoint:10
        ~output_ledger ()
    in
    let () = print_endline replayer_output in
    let output_ledger = Replayer.Output.of_json_file_exn output_ledger in
    assert (
      String.equal output_ledger.target_epoch_ledgers_state_hash
        latest_state_hash ) ;

    let%bind perf_data = extract_perf_metrics (output ^ "/archive.log") in
    perf_metrics_to_yojson perf_data |> Yojson.to_file "archive.perf" ;

    Deferred.Or_error.return Mina_automation_fixture.Intf.Passed
end

let () =
  Backtrace.elide := false ;
  Async.Scheduler.set_record_backtraces true

let () =
  let open Alcotest in
  run "Test archive node."
    [ ( "precomputed_blocks"
      , [ test_case
            "Recreate database from precomputed blocks sent from mock daemon"
            `Quick
            (Runner.run_blocking
               ( module Mina_automation_fixture.Archive.Make_FixtureWithBootstrap
                          (ArchivePrecomputedBlocksFromDaemon) ) )
        ] )
    ]

1	open Async
2	open Core
3	open Mina_automation
4	open Mina_automation_runner
5	open Mina_automation_fixture.Archive
6
7	(**
8	* Test the basic functionality of the mina archive with mocked deamon
9	*)
10
11	(* asserts count of archive blocked (we are skipping genesis block) *)
12	let assert_archived_blocks ~archive_uri ~expected =
13	let connection = Psql.Conn_str archive_uri in	1✔
14	let%bind actual_blocks_count =
15	Psql.run_command ~connection	1✔
16	"SELECT COUNT(*) FROM blocks WHERE global_slot_since_genesis > 1"
17	in
18	let actual_blocks_count =	1✔
19	match actual_blocks_count with
20	\| Ok count ->	1✔
21	count \|> Int.of_string	1✔
22	\| Error err ->	×
23	failwith ("Failed to query blocks count: " ^ Error.to_string_hum err)	×
24	in
25	if Int.( <> ) actual_blocks_count expected then
26	failwithf "Invalid number of archive blocks. Actual (%d) vs Expected (%d)"	×
27	actual_blocks_count expected ()
28	else Deferred.unit	1✔
29
30	(* Convert performance metrics to a JSON format suitable for output *)
31	(* The metrics are expected to be a list of tuples (operation, avg_time) *)
32	(* where operation is a string and avg_time is a float representing the average time in milliseconds *)
33	let perf_metrics_to_yojson metrics =
34	let json_list =	1✔
35	List.map metrics ~f:(fun (operation, avg_time) ->
36	`Assoc	10✔
37	[ ("operation", `String operation); ("avg_time_ms", `Float avg_time) ] )
38	in
39	`List json_list	1✔
40
41	(** Extract performance metrics from a log file and calculate average execution times.
42
43	This function reads a log file line by line, parses each line as a JSON log entry,
44	and extracts performance metrics identified by the "is_perf_metric" metadata field.
45	For each performance metric entry, it extracts the "elapsed" time and "label" fields.
46
47	@param log_file Path to the log file to process
48	@return A deferred list of tuples containing (operation_label, average_time_in_ms)
49
50	The function performs the following steps:
51	1. Reads all lines from the specified log file
52	2. Filters and parses lines containing performance metrics
53	3. Groups metrics by operation label
54	4. Calculates the average execution time for each operation
55
56	@raises Failure if a log line cannot be parsed as valid JSON
57	@raises exn if required metadata fields ("elapsed" or "label") are missing *)
58	let extract_perf_metrics log_file =
59	let open Deferred.Let_syntax in	1✔
60	let%bind lines = Reader.file_lines log_file in	1✔
61	let perf_metrics =	1✔
62	List.filter_map lines ~f:(fun line ->
63	if String.is_empty line then None	2,399✔
64	else
65	match Logger.Message.of_yojson (Yojson.Safe.from_string line) with	2,675✔
66	\| Ok entry ->	2,675✔
67	if String.Map.mem entry.metadata "is_perf_metric" then
68	let time_in_ms =	2,498✔
69	String.Map.find entry.metadata "elapsed"	2,498✔
70	\|> Option.value_exn	2,498✔
71	~message:
72	("Missing elapsed in log entry in log line: " ^ line)
73	\|> Yojson.Safe.Util.to_float
74	in
75	let label =	2,498✔
76	String.Map.find entry.metadata "label"	2,498✔
77	\|> Option.value_exn	2,498✔
78	~message:
79	("Missing label in log entry in log line: " ^ line)
80	\|> Yojson.Safe.Util.to_string
81	in
82	Some (label, time_in_ms)	2,498✔
83	else None	177✔
NEW 84	\| Error err ->	×
85	failwithf "Invalid log line: %s. Error: %s" line err () )
86	in
87	(* Calculate the average time for each operation *)
88	(* Group by operation and calculate the average time *)
89	let averaged_metrics =	1✔
90	String.Map.of_alist_multi perf_metrics	1✔
91	\|> Map.to_alist	1✔
92	\|> List.map ~f:(fun (operation, times) ->
93	let avg_time =	10✔
94	List.fold times ~init:0.0 ~f:( +. )	10✔
95	/. Float.of_int (List.length times)	10✔
96	in
97	(operation, avg_time) )
98	in
99	Deferred.return averaged_metrics	1✔
100
101	module ArchivePrecomputedBlocksFromDaemon = struct
102	type t = Mina_automation_fixture.Archive.after_bootstrap
103
104	let test_case (test_data : t) =
105	let open Deferred.Let_syntax in	1✔
106	let daemon = Daemon.default in
107	let archive_uri = test_data.archive.config.postgres_uri in
108	let output = test_data.temp_dir in
109	let%bind precomputed_blocks =
110	Network_data.untar_precomputed_blocks test_data.network_data output	1✔
111	in
112	let precomputed_blocks =	1✔
113	List.map precomputed_blocks ~f:(fun file -> output ^ "/" ^ file)	1✔
114	\|> List.filter ~f:(fun file -> String.is_suffix file ~suffix:".json")	34✔
115	in
116	Archive.Process.start_logging test_data.archive	1✔
117	~log_file:(output ^ "/archive.log") ;
118	let%bind () =
119	Daemon.archive_blocks_from_files daemon	1✔
120	~archive_address:test_data.archive.config.server_port
121	~format:Archive_blocks.Precomputed precomputed_blocks
122	in
123
124	let%bind () =
125	assert_archived_blocks ~archive_uri
126	~expected:(List.length precomputed_blocks)	1✔
127	in
128	let connection = Psql.Conn_str archive_uri in	1✔
129	let%bind latest_state_hash =
130	Psql.run_command ~connection	1✔
131	"SELECT state_hash FROM blocks ORDER BY id DESC LIMIT 1"
132	in
133	let latest_state_hash =	1✔
134	match latest_state_hash with
135	\| Ok hash ->	1✔
136	hash
137	\| Error err ->	×
138	failwith
139	("Failed to query latest state hash: " ^ Error.to_string_hum err)	×
140	in
141	let output_ledger = output ^ "/output_ledger.json" in
142	let replayer = Replayer.default in
143	let%bind replayer_output =
144	Replayer.run replayer ~archive_uri	1✔
145	~input_config:
146	(Network_data.replayer_input_file_path test_data.network_data)	1✔
147	~target_state_hash:latest_state_hash ~interval_checkpoint:10
148	~output_ledger ()
149	in
150	let () = print_endline replayer_output in	1✔
151	let output_ledger = Replayer.Output.of_json_file_exn output_ledger in	1✔
152	assert (	1✔
153	String.equal output_ledger.target_epoch_ledgers_state_hash	1✔
154	latest_state_hash ) ;
155
156	let%bind perf_data = extract_perf_metrics (output ^ "/archive.log") in	1✔
157	perf_metrics_to_yojson perf_data \|> Yojson.to_file "archive.perf" ;	1✔
158
159	Deferred.Or_error.return Mina_automation_fixture.Intf.Passed	1✔
160	end
161
162	let () =
163	Backtrace.elide := false ;
164	Async.Scheduler.set_record_backtraces true	1✔
165
166	let () =
167	let open Alcotest in
168	run "Test archive node."	×
169	[ ( "precomputed_blocks"
170	, [ test_case	1✔
171	"Recreate database from precomputed blocks sent from mock daemon"
172	`Quick
173	(Runner.run_blocking	1✔
174	( module Mina_automation_fixture.Archive.Make_FixtureWithBootstrap
175	(ArchivePrecomputedBlocksFromDaemon) ) )
176	] )
177	]

MinaProtocol / mina / 431

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