2863

Committed 05 Nov 2024 06:20PM UTC coverage: 30.754% (-16.6%) from 47.311%

Build # 2863

Build Type

push

buildkite

Committed by

web-flow

Commit Message

Merge pull request #16296 from MinaProtocol/dkijania/more_multi_jobs

more multi jobs in CI

Run Details

20276 of 65930 relevant lines covered (30.75%)

8631.7 hits per line

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

1.63

/src/lib/transition_frontier/persistent_root/persistent_root.ml

open Core
open Mina_base
module Ledger = Mina_ledger.Ledger
open Frontier_base
module Ledger_transfer = Mina_ledger.Ledger_transfer.Make (Ledger) (Ledger.Db)

let genesis_root_identifier ~genesis_state_hash =
  let open Root_identifier.Stable.Latest in
  { state_hash = genesis_state_hash }

let with_file ?size filename access_level ~f =
  let open Unix in
  let shared, mode =
    match access_level with
    | `Read ->
        (false, [ O_RDONLY ])
    | `Write ->
        (true, [ O_RDWR; O_TRUNC; O_CREAT ])
  in
  let fd = Unix.openfile filename ~mode in
  let buf_size =
    match size with
    | None ->
        Int64.to_int_exn Unix.(fstat fd).st_size
    | Some sz ->
        sz
  in
  (* Bigstring.map_file has been removed. We copy its old implementation. *)
  let buf =
    Bigarray.(
      array1_of_genarray
        (Core.Unix.map_file fd char c_layout ~shared [| buf_size |]))
  in
  let x = f buf in
  Bigstring.unsafe_destroy buf ;
  Unix.close fd ;
  x

module Locations = struct
  let snarked_ledger root = Filename.concat root "snarked_ledger"

  let tmp_snarked_ledger root = Filename.concat root "tmp_snarked_ledger"

  (** potential_snarked_ledgers is a json file that stores a list of potential
      snarked ledgeres *)
  let potential_snarked_ledgers root =
    Filename.concat root "potential_snarked_ledgers.json"

  (** potential_snarked_ledger is the actual location of each potential snarked
      ledger *)
  let potential_snarked_ledger root =
    let uuid = Uuid_unix.create () in
    Filename.concat root ("snarked_ledger" ^ Uuid.to_string_hum uuid)

  let root_identifier root = Filename.concat root "root"
end

(* TODO: create a reusable singleton factory abstraction *)
module rec Instance_type : sig
  type t =
    { snarked_ledger : Ledger.Db.t
    ; potential_snarked_ledgers : string Queue.t
    ; factory : Factory_type.t
    }
end =
  Instance_type

and Factory_type : sig
  type t =
    { directory : string
    ; logger : Logger.t
    ; mutable instance : Instance_type.t option
    ; ledger_depth : int
    }
end =
  Factory_type

open Instance_type
open Factory_type

module Instance = struct
  type t = Instance_type.t

  let potential_snarked_ledgers_to_yojson queue =
    `List (List.map (Queue.to_list queue) ~f:(fun filename -> `String filename))

  let potential_snarked_ledgers_of_yojson json =
    Yojson.Safe.Util.to_list json |> List.map ~f:Yojson.Safe.Util.to_string

  let load_potential_snarked_ledgers_from_disk factory =
    let location = Locations.potential_snarked_ledgers factory.directory in
    if phys_equal (Sys.file_exists location) `Yes then
      Yojson.Safe.from_file location |> potential_snarked_ledgers_of_yojson
    else []

  let write_potential_snarked_ledgers_to_disk t =
    Yojson.Safe.to_file
      (Locations.potential_snarked_ledgers t.factory.directory)
      (potential_snarked_ledgers_to_yojson t.potential_snarked_ledgers)

  let enqueue_snarked_ledger ~location t =
    Queue.enqueue t.potential_snarked_ledgers location ;
    write_potential_snarked_ledgers_to_disk t

  let dequeue_snarked_ledger t =
    let location = Queue.dequeue_exn t.potential_snarked_ledgers in
    File_system.rmrf location ;
    write_potential_snarked_ledgers_to_disk t

  let destroy t =
    List.iter (Queue.to_list t.potential_snarked_ledgers) ~f:File_system.rmrf ;
    File_system.rmrf (Locations.potential_snarked_ledgers t.factory.directory) ;
    Ledger.Db.close t.snarked_ledger ;
    t.factory.instance <- None

  let close t =
    Ledger.Db.close t.snarked_ledger ;
    t.factory.instance <- None

  let create factory =
    let snarked_ledger =
      Ledger.Db.create ~depth:factory.ledger_depth
        ~directory_name:(Locations.snarked_ledger factory.directory)
        ()
    in
    { snarked_ledger; potential_snarked_ledgers = Queue.create (); factory }

  (** When we load from disk,
      1. Check the potential_snarked_ledgers to see if any one of these
         matches the snarked_ledger_hash in persistent_frontier;
      2. if none of those works, we load the old snarked_ledger and check if
         the old snarked_ledger matches with persistent_frontier;
      3. if not, we just reset all the persisted data and start from genesis
   *)
  let load_from_disk factory ~snarked_ledger_hash ~logger =
    let potential_snarked_ledgers =
      load_potential_snarked_ledgers_from_disk factory
    in
    let snarked_ledger =
      List.fold_until potential_snarked_ledgers ~init:None
        ~f:(fun _ location ->
          let potential_snarked_ledger =
            Ledger.Db.create ~depth:factory.ledger_depth
              ~directory_name:location ()
          in
          let potential_snarked_ledger_hash =
            Frozen_ledger_hash.of_ledger_hash
            @@ Ledger.Db.merkle_root potential_snarked_ledger
          in
          [%log debug]
            ~metadata:
              [ ( "potential_snarked_ledger_hash"
                , Frozen_ledger_hash.to_yojson potential_snarked_ledger_hash )
              ]
            "loaded potential_snarked_ledger from disk" ;
          if
            Frozen_ledger_hash.equal potential_snarked_ledger_hash
              snarked_ledger_hash
          then (
            let snarked_ledger =
              Ledger.Db.create ~depth:factory.ledger_depth
                ~directory_name:(Locations.tmp_snarked_ledger factory.directory)
                ()
            in
            match
              Ledger_transfer.transfer_accounts
                ~src:(Ledger.of_database potential_snarked_ledger)
                ~dest:snarked_ledger
            with
            | Ok _ ->
                Ledger.Db.close potential_snarked_ledger ;
                File_system.rmrf @@ Locations.snarked_ledger factory.directory ;
                Sys.rename
                  (Locations.tmp_snarked_ledger factory.directory)
                  (Locations.snarked_ledger factory.directory) ;
                List.iter potential_snarked_ledgers ~f:File_system.rmrf ;
                File_system.rmrf
                  (Locations.potential_snarked_ledgers factory.directory) ;
                Stop (Some snarked_ledger)
            | Error e ->
                Ledger.Db.close potential_snarked_ledger ;
                List.iter potential_snarked_ledgers ~f:File_system.rmrf ;
                File_system.rmrf
                  (Locations.potential_snarked_ledgers factory.directory) ;
                [%log' error factory.logger]
                  ~metadata:[ ("error", `String (Error.to_string_hum e)) ]
                  "Ledger_transfer failed" ;
                Stop None )
          else (
            Ledger.Db.close potential_snarked_ledger ;
            Continue None ) )
        ~finish:(fun _ ->
          List.iter potential_snarked_ledgers ~f:File_system.rmrf ;
          File_system.rmrf
            (Locations.potential_snarked_ledgers factory.directory) ;
          None )
    in
    match snarked_ledger with
    | None ->
        let snarked_ledger =
          Ledger.Db.create ~depth:factory.ledger_depth
            ~directory_name:(Locations.snarked_ledger factory.directory)
            ()
        in
        let potential_snarked_ledger_hash =
          Frozen_ledger_hash.of_ledger_hash
          @@ Ledger.Db.merkle_root snarked_ledger
        in
        if
          Frozen_ledger_hash.equal potential_snarked_ledger_hash
            snarked_ledger_hash
        then
          Ok
            { snarked_ledger
            ; potential_snarked_ledgers = Queue.create ()
            ; factory
            }
        else (
          Ledger.Db.close snarked_ledger ;
          Error `Snarked_ledger_mismatch )
    | Some snarked_ledger ->
        Ok
          { snarked_ledger
          ; potential_snarked_ledgers = Queue.create ()
          ; factory
          }

  (* TODO: encapsulate functionality of snarked ledger *)
  let snarked_ledger { snarked_ledger; _ } = snarked_ledger

  let set_root_identifier t new_root_identifier =
    [%log' trace t.factory.logger]
      ~metadata:
        [ ("root_identifier", Root_identifier.to_yojson new_root_identifier) ]
      "Setting persistent root identifier" ;
    let size = Root_identifier.Stable.Latest.bin_size_t new_root_identifier in
    with_file (Locations.root_identifier t.factory.directory) `Write ~size
      ~f:(fun buf ->
        ignore
          ( Root_identifier.Stable.Latest.bin_write_t buf ~pos:0
              new_root_identifier
            : int ) )

  (* defaults to genesis *)
  let load_root_identifier t =
    let file = Locations.root_identifier t.factory.directory in
    match Unix.access file [ `Exists; `Read ] with
    | Error _ ->
        None
    | Ok () ->
        with_file file `Read ~f:(fun buf ->
            let root_identifier =
              Root_identifier.Stable.Latest.bin_read_t buf ~pos_ref:(ref 0)
            in
            [%log' trace t.factory.logger]
              ~metadata:
                [ ("root_identifier", Root_identifier.to_yojson root_identifier)
                ]
              "Loaded persistent root identifier" ;
            Some root_identifier )

  let set_root_state_hash t state_hash = set_root_identifier t { state_hash }
end

type t = Factory_type.t

let create ~logger ~directory ~ledger_depth =
  { directory; logger; instance = None; ledger_depth }

let create_instance_exn t =
  assert (Option.is_none t.instance) ;
  let instance = Instance.create t in
  t.instance <- Some instance ;
  instance

let load_from_disk_exn t ~snarked_ledger_hash ~logger =
  let open Result.Let_syntax in
  assert (Option.is_none t.instance) ;
  let%map instance = Instance.load_from_disk t ~snarked_ledger_hash ~logger in
  t.instance <- Some instance ;
  instance

let with_instance_exn t ~f =
  let instance = create_instance_exn t in
  let x = f instance in
  Instance.close instance ; x

let reset_to_genesis_exn t ~precomputed_values =
  assert (Option.is_none t.instance) ;
  File_system.rmrf t.directory ;
  with_instance_exn t ~f:(fun instance ->
      ignore
        ( Ledger_transfer.transfer_accounts
            ~src:
              (Lazy.force
                 (Precomputed_values.genesis_ledger precomputed_values) )
            ~dest:(Instance.snarked_ledger instance)
          : Ledger.Db.t Or_error.t ) ;
      Instance.set_root_identifier instance
        (genesis_root_identifier
           ~genesis_state_hash:
             (Precomputed_values.genesis_state_hashes precomputed_values)
               .state_hash ) )

1	open Core	3✔
2	open Mina_base
3	module Ledger = Mina_ledger.Ledger
4	open Frontier_base
5	module Ledger_transfer = Mina_ledger.Ledger_transfer.Make (Ledger) (Ledger.Db)
6
7	let genesis_root_identifier ~genesis_state_hash =
8	let open Root_identifier.Stable.Latest in	×
9	{ state_hash = genesis_state_hash }
10
11	let with_file ?size filename access_level ~f =
12	let open Unix in	×
13	let shared, mode =
14	match access_level with
15	\| `Read ->	×
16	(false, [ O_RDONLY ])
17	\| `Write ->	×
18	(true, [ O_RDWR; O_TRUNC; O_CREAT ])
19	in
20	let fd = Unix.openfile filename ~mode in
21	let buf_size =	×
22	match size with
23	\| None ->	×
24	Int64.to_int_exn Unix.(fstat fd).st_size	×
25	\| Some sz ->	×
26	sz
27	in
28	(* Bigstring.map_file has been removed. We copy its old implementation. *)
29	let buf =
30	Bigarray.(
31	array1_of_genarray	×
32	(Core.Unix.map_file fd char c_layout ~shared [\| buf_size \|]))	×
33	in
34	let x = f buf in
35	Bigstring.unsafe_destroy buf ;	×
36	Unix.close fd ;	×
37	x	×
38
39	module Locations = struct
40	let snarked_ledger root = Filename.concat root "snarked_ledger"	×
41
42	let tmp_snarked_ledger root = Filename.concat root "tmp_snarked_ledger"	×
43
44	(** potential_snarked_ledgers is a json file that stores a list of potential
45	snarked ledgeres *)
46	let potential_snarked_ledgers root =
47	Filename.concat root "potential_snarked_ledgers.json"	×
48
49	(** potential_snarked_ledger is the actual location of each potential snarked
50	ledger *)
51	let potential_snarked_ledger root =
52	let uuid = Uuid_unix.create () in	×
53	Filename.concat root ("snarked_ledger" ^ Uuid.to_string_hum uuid)	×
54
55	let root_identifier root = Filename.concat root "root"	×
56	end
57
58	(* TODO: create a reusable singleton factory abstraction *)
59	module rec Instance_type : sig
60	type t =
61	{ snarked_ledger : Ledger.Db.t
62	; potential_snarked_ledgers : string Queue.t
63	; factory : Factory_type.t
64	}
65	end =
66	Instance_type
67
68	and Factory_type : sig
69	type t =
70	{ directory : string
71	; logger : Logger.t
72	; mutable instance : Instance_type.t option
73	; ledger_depth : int
74	}
75	end =
76	Factory_type
77
78	open Instance_type
79	open Factory_type
80
81	module Instance = struct
82	type t = Instance_type.t
83
84	let potential_snarked_ledgers_to_yojson queue =
85	`List (List.map (Queue.to_list queue) ~f:(fun filename -> `String filename))	×
86
87	let potential_snarked_ledgers_of_yojson json =
88	Yojson.Safe.Util.to_list json \|> List.map ~f:Yojson.Safe.Util.to_string	×
89
90	let load_potential_snarked_ledgers_from_disk factory =
91	let location = Locations.potential_snarked_ledgers factory.directory in	×
92	if phys_equal (Sys.file_exists location) `Yes then	×
93	Yojson.Safe.from_file location \|> potential_snarked_ledgers_of_yojson	×
94	else []	×
95
96	let write_potential_snarked_ledgers_to_disk t =
97	Yojson.Safe.to_file	×
98	(Locations.potential_snarked_ledgers t.factory.directory)	×
99	(potential_snarked_ledgers_to_yojson t.potential_snarked_ledgers)	×
100
101	let enqueue_snarked_ledger ~location t =
102	Queue.enqueue t.potential_snarked_ledgers location ;	×
103	write_potential_snarked_ledgers_to_disk t	×
104
105	let dequeue_snarked_ledger t =
106	let location = Queue.dequeue_exn t.potential_snarked_ledgers in	×
107	File_system.rmrf location ;	×
108	write_potential_snarked_ledgers_to_disk t	×
109
110	let destroy t =
111	List.iter (Queue.to_list t.potential_snarked_ledgers) ~f:File_system.rmrf ;	×
112	File_system.rmrf (Locations.potential_snarked_ledgers t.factory.directory) ;	×
113	Ledger.Db.close t.snarked_ledger ;	×
114	t.factory.instance <- None	×
115
116	let close t =
117	Ledger.Db.close t.snarked_ledger ;	×
118	t.factory.instance <- None	×
119
120	let create factory =
121	let snarked_ledger =	×
122	Ledger.Db.create ~depth:factory.ledger_depth
123	~directory_name:(Locations.snarked_ledger factory.directory)	×
124	()
125	in
126	{ snarked_ledger; potential_snarked_ledgers = Queue.create (); factory }	×
127
128	(** When we load from disk,
129	1. Check the potential_snarked_ledgers to see if any one of these
130	matches the snarked_ledger_hash in persistent_frontier;
131	2. if none of those works, we load the old snarked_ledger and check if
132	the old snarked_ledger matches with persistent_frontier;
133	3. if not, we just reset all the persisted data and start from genesis
134	*)
135	let load_from_disk factory ~snarked_ledger_hash ~logger =
136	let potential_snarked_ledgers =	×
137	load_potential_snarked_ledgers_from_disk factory
138	in
139	let snarked_ledger =	×
140	List.fold_until potential_snarked_ledgers ~init:None
141	~f:(fun _ location ->
142	let potential_snarked_ledger =	×
143	Ledger.Db.create ~depth:factory.ledger_depth
144	~directory_name:location ()
145	in
146	let potential_snarked_ledger_hash =	×
147	Frozen_ledger_hash.of_ledger_hash
148	@@ Ledger.Db.merkle_root potential_snarked_ledger	×
149	in
150	[%log debug]	×
151	~metadata:
152	[ ( "potential_snarked_ledger_hash"
153	, Frozen_ledger_hash.to_yojson potential_snarked_ledger_hash )	×
154	]
155	"loaded potential_snarked_ledger from disk" ;
156	if	×
157	Frozen_ledger_hash.equal potential_snarked_ledger_hash
158	snarked_ledger_hash
159	then (	×
160	let snarked_ledger =
161	Ledger.Db.create ~depth:factory.ledger_depth
162	~directory_name:(Locations.tmp_snarked_ledger factory.directory)	×
163	()
164	in
165	match	×
166	Ledger_transfer.transfer_accounts
167	~src:(Ledger.of_database potential_snarked_ledger)	×
168	~dest:snarked_ledger
169	with
170	\| Ok _ ->	×
171	Ledger.Db.close potential_snarked_ledger ;
172	File_system.rmrf @@ Locations.snarked_ledger factory.directory ;	×
173	Sys.rename	×
174	(Locations.tmp_snarked_ledger factory.directory)	×
175	(Locations.snarked_ledger factory.directory) ;	×
176	List.iter potential_snarked_ledgers ~f:File_system.rmrf ;	×
177	File_system.rmrf	×
178	(Locations.potential_snarked_ledgers factory.directory) ;	×
179	Stop (Some snarked_ledger)	×
180	\| Error e ->	×
181	Ledger.Db.close potential_snarked_ledger ;
182	List.iter potential_snarked_ledgers ~f:File_system.rmrf ;	×
183	File_system.rmrf	×
184	(Locations.potential_snarked_ledgers factory.directory) ;	×
185	[%log' error factory.logger]	×
186	~metadata:[ ("error", `String (Error.to_string_hum e)) ]	×
187	"Ledger_transfer failed" ;
188	Stop None )	×
189	else (	×
190	Ledger.Db.close potential_snarked_ledger ;
191	Continue None ) )	×
192	~finish:(fun _ ->
193	List.iter potential_snarked_ledgers ~f:File_system.rmrf ;	×
194	File_system.rmrf	×
195	(Locations.potential_snarked_ledgers factory.directory) ;	×
196	None )	×
197	in
198	match snarked_ledger with	×
199	\| None ->	×
200	let snarked_ledger =
201	Ledger.Db.create ~depth:factory.ledger_depth
202	~directory_name:(Locations.snarked_ledger factory.directory)	×
203	()
204	in
205	let potential_snarked_ledger_hash =	×
206	Frozen_ledger_hash.of_ledger_hash
207	@@ Ledger.Db.merkle_root snarked_ledger	×
208	in
209	if	×
210	Frozen_ledger_hash.equal potential_snarked_ledger_hash
211	snarked_ledger_hash
212	then
213	Ok	×
214	{ snarked_ledger
215	; potential_snarked_ledgers = Queue.create ()	×
216	; factory
217	}
218	else (	×
219	Ledger.Db.close snarked_ledger ;
220	Error `Snarked_ledger_mismatch )	×
221	\| Some snarked_ledger ->	×
222	Ok
223	{ snarked_ledger
224	; potential_snarked_ledgers = Queue.create ()	×
225	; factory
226	}
227
228	(* TODO: encapsulate functionality of snarked ledger *)
229	let snarked_ledger { snarked_ledger; _ } = snarked_ledger	×
230
231	let set_root_identifier t new_root_identifier =
232	[%log' trace t.factory.logger]	×
233	~metadata:
234	[ ("root_identifier", Root_identifier.to_yojson new_root_identifier) ]	×
235	"Setting persistent root identifier" ;
236	let size = Root_identifier.Stable.Latest.bin_size_t new_root_identifier in	×
237	with_file (Locations.root_identifier t.factory.directory) `Write ~size	×
238	~f:(fun buf ->
239	ignore	×
240	( Root_identifier.Stable.Latest.bin_write_t buf ~pos:0	×
241	new_root_identifier
242	: int ) )
243
244	(* defaults to genesis *)
245	let load_root_identifier t =
246	let file = Locations.root_identifier t.factory.directory in	×
247	match Unix.access file [ `Exists; `Read ] with	×
248	\| Error _ ->	×
249	None
250	\| Ok () ->	×
251	with_file file `Read ~f:(fun buf ->
252	let root_identifier =	×
253	Root_identifier.Stable.Latest.bin_read_t buf ~pos_ref:(ref 0)
254	in
255	[%log' trace t.factory.logger]	×
256	~metadata:
257	[ ("root_identifier", Root_identifier.to_yojson root_identifier)	×
258	]
259	"Loaded persistent root identifier" ;
260	Some root_identifier )	×
261
262	let set_root_state_hash t state_hash = set_root_identifier t { state_hash }	×
263	end
264
265	type t = Factory_type.t
266
267	let create ~logger ~directory ~ledger_depth =
268	{ directory; logger; instance = None; ledger_depth }	×
269
270	let create_instance_exn t =
271	assert (Option.is_none t.instance) ;	×
272	let instance = Instance.create t in
273	t.instance <- Some instance ;	×
274	instance
275
276	let load_from_disk_exn t ~snarked_ledger_hash ~logger =
277	let open Result.Let_syntax in	×
278	assert (Option.is_none t.instance) ;	×
279	let%map instance = Instance.load_from_disk t ~snarked_ledger_hash ~logger in	×
280	t.instance <- Some instance ;	×
281	instance
282
283	let with_instance_exn t ~f =
284	let instance = create_instance_exn t in	×
285	let x = f instance in	×
286	Instance.close instance ; x	×
287
288	let reset_to_genesis_exn t ~precomputed_values =
289	assert (Option.is_none t.instance) ;	×
290	File_system.rmrf t.directory ;
291	with_instance_exn t ~f:(fun instance ->	×
292	ignore	×
293	( Ledger_transfer.transfer_accounts
294	~src:
295	(Lazy.force	×
296	(Precomputed_values.genesis_ledger precomputed_values) )	×
297	~dest:(Instance.snarked_ledger instance)	×
298	: Ledger.Db.t Or_error.t ) ;
299	Instance.set_root_identifier instance
300	(genesis_root_identifier
301	~genesis_state_hash:
302	(Precomputed_values.genesis_state_hashes precomputed_values)	×
303	.state_hash ) )	6✔

MinaProtocol / mina / 2863

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