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

0.73

/src/lib/network_pool/batcher.ml

open Core_kernel
open Async_kernel
open Network_peer

(* Only show stdout for failed inline tests. *)
open Inline_test_quiet_logs

module Id = Unique_id.Int ()

type ('init, 'result) elt =
  { id : Id.t
  ; data : 'init
  ; weight : int
  ; res : (('result, Verifier.invalid) Result.t Or_error.t Ivar.t[@sexp.opaque])
  }
[@@deriving sexp]

type ('proof, 'result) state =
  | Waiting
  | Verifying of { out_for_verification : ('proof, 'result) elt list }
[@@deriving sexp]

module Q = Doubly_linked

type ('init, 'partially_validated, 'result) t =
  { mutable state : ('init, 'result) state
  ; how_to_add : [ `Insert | `Enqueue_back ]
  ; queue : ('init, 'result) elt Q.t
  ; compare_init : ('init -> 'init -> int) option
  ; logger : (Logger.t[@sexp.opaque])
  ; weight : 'init -> int
  ; max_weight_per_call : int option
  ; verifier :
      (   (* The batched verifier may make partial progress on its input so that we can
             save time when it is re-verified in a smaller batch in the case that a batch
             fails to verify.
          *)
          [ `Init of 'init | `Partially_validated of 'partially_validated ] list
       -> [ `Valid of 'result
          | Verifier.invalid
          | `Potentially_invalid of 'partially_validated * Error.t ]
          list
          Deferred.Or_error.t
      [@sexp.opaque] )
  }
[@@deriving sexp]

let create ?(how_to_add = `Enqueue_back) ?logger ?compare_init
    ?(weight = fun _ -> 1) ?max_weight_per_call verifier =
  { state = Waiting
  ; queue = Q.create ()
  ; how_to_add
  ; compare_init
  ; verifier
  ; weight
  ; max_weight_per_call
  ; logger = Option.value logger ~default:(Logger.create ())
  }

let call_verifier t (ps : 'proof list) = t.verifier ps

(* Worst case (if all the proofs are invalid): log n * (2^(log n) + 1)
   In the average case this should show better performance.
   We could implement the trusted/untrusted batches from the snark pool batching RFC #4882
   to further mitigate possible DoS/DDoS here
*)
let rec determine_outcome :
    type p r partial.
       (p, r) elt list
    -> [ `Valid of r
       | `Potentially_invalid of partial * Error.t
       | Verifier.invalid ]
       list
    -> (p, partial, r) t
    -> unit Deferred.Or_error.t =
 fun ps res v ->
  O1trace.thread "determining_batcher_outcome" (fun () ->
      (* First separate out all the known results. That information will definitely be included
         in the outcome.
      *)
      let logger = Logger.create () in
      let potentially_invalid =
        List.filter_map (List.zip_exn ps res) ~f:(fun (elt, r) ->
            match r with
            | `Valid r ->
                if Ivar.is_full elt.res then
                  [%log error] "Ivar.fill bug is here!" ;
                Ivar.fill elt.res (Ok (Ok r)) ;
                None
            | `Invalid_keys keys ->
                if Ivar.is_full elt.res then
                  [%log error] "Ivar.fill bug is here!" ;
                Ivar.fill elt.res (Ok (Error (`Invalid_keys keys))) ;
                None
            | `Invalid_signature keys ->
                if Ivar.is_full elt.res then
                  [%log error] "Ivar.fill bug is here!" ;
                Ivar.fill elt.res (Ok (Error (`Invalid_signature keys))) ;
                None
            | `Invalid_proof err ->
                if Ivar.is_full elt.res then
                  [%log error] "Ivar.fill bug is here!" ;
                Ivar.fill elt.res (Ok (Error (`Invalid_proof err))) ;
                None
            | `Missing_verification_key keys ->
                if Ivar.is_full elt.res then
                  [%log error] "Ivar.fill bug is here!" ;
                Ivar.fill elt.res (Ok (Error (`Missing_verification_key keys))) ;
                None
            | `Unexpected_verification_key keys ->
                if Ivar.is_full elt.res then
                  [%log error] "Ivar.fill bug is here!" ;
                Ivar.fill elt.res
                  (Ok (Error (`Unexpected_verification_key keys))) ;
                None
            | `Mismatched_authorization_kind keys ->
                if Ivar.is_full elt.res then
                  [%log error] "Ivar.fill bug is here!" ;
                Ivar.fill elt.res
                  (Ok (Error (`Mismatched_authorization_kind keys))) ;
                None
            | `Potentially_invalid (new_hint, err) ->
                Some (elt, new_hint, err) )
      in
      let open Deferred.Or_error.Let_syntax in
      match potentially_invalid with
      | [] ->
          (* All results are known *)
          return ()
      | [ ({ res; _ }, _, err) ] ->
          if Ivar.is_full res then [%log error] "Ivar.fill bug is here!" ;
          Ivar.fill res (Ok (Error (`Invalid_proof err))) ;
          (* If there is a potentially invalid proof in this batch of size 1, then
             that proof is itself invalid. *)
          return ()
      | _ ->
          let outcome xs =
            let%bind res_xs =
              call_verifier v
                (List.map xs ~f:(fun (_e, new_hint, _) ->
                     `Partially_validated new_hint ) )
            in
            determine_outcome (List.map xs ~f:(fun (e, _, _) -> e)) res_xs v
          in
          let length = List.length potentially_invalid in
          let left, right = List.split_n potentially_invalid (length / 2) in
          let%bind () = outcome left in
          outcome right )

let compare_elt ~compare t1 t2 =
  match compare t1.data t2.data with 0 -> Id.compare t1.id t2.id | x -> x

let order_proofs t =
  match t.compare_init with
  | None ->
      Fn.id
  | Some compare ->
      List.sort ~compare:(compare_elt ~compare)

(* When new proofs come in put them in the queue.
      If state = Waiting, verify those proofs immediately.
      Whenever the verifier returns, if the queue is nonempty, flush it into the verifier.
*)

let rec start_verifier : type proof partial r. (proof, partial, r) t -> unit =
 fun t ->
  O1trace.sync_thread "running_batcher_verifier_loop" (fun () ->
      if Q.is_empty t.queue then
        (* we looped in the else after verifier finished but no pending work. *)
        t.state <- Waiting
      else (
        [%log' debug t.logger] "Verifying proofs in batch of size $num_proofs"
          ~metadata:[ ("num_proofs", `Int (Q.length t.queue)) ] ;
        let out_for_verification =
          let proofs =
            match t.max_weight_per_call with
            | None ->
                let proofs = Q.to_list t.queue in
                Q.clear t.queue ; proofs
            | Some max_weight ->
                let rec take capacity acc =
                  match Q.first t.queue with
                  | None ->
                      acc
                  | Some ({ weight; _ } as proof) ->
                      if weight <= capacity then (
                        ignore (Q.remove_first t.queue : (proof, r) elt option) ;
                        take (capacity - weight) (proof :: acc) )
                      else acc
                in
                List.rev (take max_weight [])
          in
          order_proofs t proofs
        in
        [%log' debug t.logger] "Calling verifier with $num_proofs on $ids"
          ~metadata:
            [ ("num_proofs", `Int (List.length out_for_verification))
            ; ( "ids"
              , `List
                  (List.map
                     ~f:(fun { id; _ } -> `Int (Id.to_int_exn id))
                     out_for_verification ) )
            ] ;
        let res =
          match%bind
            call_verifier t
              (List.map out_for_verification ~f:(fun { data = p; _ } -> `Init p))
          with
          | Error e ->
              Deferred.return (Error e)
          | Ok res ->
              determine_outcome out_for_verification res t
        in
        t.state <- Verifying { out_for_verification } ;
        upon res (fun r ->
            ( match r with
            | Ok () ->
                ()
            | Error e ->
                List.iter out_for_verification ~f:(fun x ->
                    Ivar.fill_if_empty x.res (Error e) ) ) ;
            start_verifier t ) ) )

let verify (type p r partial) (t : (p, partial, r) t) (proof : p) :
    (r, Verifier.invalid) Result.t Deferred.Or_error.t =
  let elt =
    { id = Id.create ()
    ; data = proof
    ; weight = t.weight proof
    ; res = Ivar.create ()
    }
  in
  ignore
    ( match (t.how_to_add, t.compare_init) with
      | `Enqueue_back, _ | `Insert, None ->
          Q.insert_last t.queue elt
      | `Insert, Some compare -> (
          (* Find the first element that [proof] is less than *)
          let compare = compare_elt ~compare in
          match Q.find_elt t.queue ~f:(fun e -> compare elt e < 0) with
          | None ->
              (* [proof] is greater than all elts in the queue, and so goes in the back. *)
              Q.insert_last t.queue elt
          | Some succ ->
              Q.insert_before t.queue succ elt )
      : (p, r) elt Q.Elt.t ) ;
  (match t.state with Verifying _ -> () | Waiting -> start_verifier t) ;
  Ivar.read elt.res

type ('a, 'b, 'c) batcher = ('a, 'b, 'c) t [@@deriving sexp]

let compare_envelope (e1 : _ Envelope.Incoming.t) (e2 : _ Envelope.Incoming.t) =
  Envelope.Sender.compare e1.sender e2.sender

module Transaction_pool = struct
  open Mina_base

  type diff = User_command.Verifiable.t list Envelope.Incoming.t
  [@@deriving sexp]

  (* A partially verified transaction is either valid, or valid assuming that some list of
     (verification key, statement, proof) triples will verify. That is, the transaction has
     already been validated in all ways, except the proofs were in a batch that failed to
     verify.
  *)
  type partial_item =
    [ `Valid of User_command.Valid.t
    | `Valid_assuming of
      User_command.Verifiable.t
      * ( Pickles.Side_loaded.Verification_key.t
        * Zkapp_statement.t
        * Pickles.Side_loaded.Proof.t )
        list ]
  [@@deriving sexp]

  type partial = partial_item list [@@deriving sexp]

  type t = (diff, partial, User_command.Valid.t list) batcher [@@deriving sexp]

  type input = [ `Init of diff | `Partially_validated of partial ]

  let init_result (ds : input list) =
    (* We store a result for every diff in the input. *)
    Array.of_list_map ds ~f:(function
      | `Init d ->
          (* Initially, the status of all the transactions in a never-before-seen
             diff are unknown. *)
          `In_progress (Array.of_list_map d.data ~f:(fun _ -> `Unknown))
      | `Partially_validated d ->
          (* We've seen this diff before, so we have some information about its
             transactions. *)
          `In_progress
            (Array.of_list_map d ~f:(function
              | `Valid c ->
                  `Valid c
              | `Valid_assuming x ->
                  `Valid_assuming x ) ) )

  let list_of_array_map a ~f = List.init (Array.length a) ~f:(fun i -> f a.(i))

  let all_valid a =
    Option.all
      (Array.to_list
         (Array.map a ~f:(function `Valid c -> Some c | _ -> None)) )

  let create verifier : t =
    let logger = Logger.create () in
    create ~compare_init:compare_envelope ~logger (fun (ds : input list) ->
        O1trace.thread "dispatching_transaction_pool_batcher_verification"
          (fun () ->
            [%log debug]
              "Dispatching $num_proofs transaction pool proofs to verifier"
              ~metadata:[ ("num_proofs", `Int (List.length ds)) ] ;
            let open Deferred.Or_error.Let_syntax in
            let result = init_result ds in
            (* Extract all the transactions that have not yet been fully validated and hold on to their
               position (diff index, position in diff). *)
            let unknowns =
              List.concat_mapi ds ~f:(fun i x ->
                  match x with
                  | `Init diff ->
                      List.mapi diff.data ~f:(fun j c -> ((i, j), c))
                  | `Partially_validated partial ->
                      List.filter_mapi partial ~f:(fun j c ->
                          match c with
                          | `Valid _ ->
                              None
                          | `Valid_assuming (v, _) ->
                              (* TODO: This rechecks the signatures on zkApp transactions... oh well for now *)
                              Some ((i, j), v) ) )
            in
            let%map res =
              (* Verify the unknowns *)
              Verifier.verify_commands verifier
                (List.map unknowns ~f:(fun (_, txn) ->
                     { With_status.data = txn; status = Applied } ) )
            in
            (* We now iterate over the results of the unknown transactions and appropriately modify
               the verification result of the diff that it belongs to. *)
            List.iter2_exn unknowns res ~f:(fun ((i, j), v) r ->
                match r with
                | `Invalid_keys keys ->
                    (* A diff is invalid is any of the transactions it contains are invalid.
                       Invalidate the whole diff that this transaction comes from. *)
                    result.(i) <- `Invalid_keys keys
                | `Invalid_signature keys ->
                    (* Invalidate the whole diff *)
                    result.(i) <- `Invalid_signature keys
                | `Missing_verification_key keys ->
                    (* Invalidate the whole diff *)
                    result.(i) <- `Missing_verification_key keys
                | `Unexpected_verification_key keys ->
                    (* Invalidate the whole diff *)
                    result.(i) <- `Unexpected_verification_key keys
                | `Mismatched_authorization_kind keys ->
                    (* Invalidate the whole diff *)
                    result.(i) <- `Mismatched_authorization_kind keys
                | `Invalid_proof err ->
                    (* Invalidate the whole diff *)
                    result.(i) <- `Invalid_proof err
                | `Valid_assuming xs -> (
                    match result.(i) with
                    | `Invalid_keys _
                    | `Invalid_signature _
                    | `Invalid_proof _
                    | `Missing_verification_key _
                    | `Unexpected_verification_key _
                    | `Mismatched_authorization_kind _ ->
                        (* If this diff has already been declared invalid, knowing that one of its
                           transactions is partially valid is not useful. *)
                        ()
                    | `In_progress a ->
                        (* The diff may still be valid. *)
                        a.(j) <- `Valid_assuming (v, xs) )
                | `Valid c -> (
                    (* Similar to the above. *)
                    match result.(i) with
                    | `Invalid_keys _
                    | `Invalid_signature _
                    | `Invalid_proof _
                    | `Missing_verification_key _
                    | `Unexpected_verification_key _
                    | `Mismatched_authorization_kind _ ->
                        ()
                    | `In_progress a ->
                        a.(j) <- `Valid c ) ) ;
            list_of_array_map result ~f:(function
              | `Invalid_keys keys ->
                  `Invalid_keys keys
              | `Invalid_signature keys ->
                  `Invalid_signature keys
              | `Invalid_proof err ->
                  `Invalid_proof err
              | `Missing_verification_key keys ->
                  `Missing_verification_key keys
              | `Unexpected_verification_key keys ->
                  `Unexpected_verification_key keys
              | `Mismatched_authorization_kind keys ->
                  `Mismatched_authorization_kind keys
              | `In_progress a -> (
                  (* If the diff is all valid, we're done. If not, we return a partial
                       result. *)
                  match all_valid a with
                  | Some res ->
                      `Valid res
                  | None ->
                      `Potentially_invalid
                        ( list_of_array_map a ~f:(function
                            | `Unknown ->
                                assert false
                            | `Valid c ->
                                `Valid c
                            | `Valid_assuming (v, xs) ->
                                `Valid_assuming (v, xs) )
                        , Error.of_string "In progress" ) ) ) ) )

  let verify (t : t) = verify t
end

module Snark_pool = struct
  type proof_envelope =
    (Ledger_proof.t One_or_two.t * Mina_base.Sok_message.t) Envelope.Incoming.t
  [@@deriving sexp]

  (* We don't use partial verification here. *)
  type partial = proof_envelope [@@deriving sexp]

  type t = (proof_envelope, partial, unit) batcher [@@deriving sexp]

  let verify (t : t) (p : proof_envelope) : bool Deferred.Or_error.t =
    let open Deferred.Or_error.Let_syntax in
    match%map verify t p with Ok () -> true | Error _ -> false

  let create verifier : t =
    let logger = Logger.create () in
    create
    (* TODO: Make this a proper config detail once we have data on what a
           good default would be.
    *)
      ~max_weight_per_call:
        (Option.value_map ~default:1000 ~f:Int.of_string
           (Sys.getenv_opt "MAX_VERIFIER_BATCH_SIZE") )
      ~compare_init:compare_envelope ~logger
      (fun ps0 ->
        [%log debug] "Dispatching $num_proofs snark pool proofs to verifier"
          ~metadata:[ ("num_proofs", `Int (List.length ps0)) ] ;
        let ps =
          List.concat_map ps0 ~f:(function
              | `Partially_validated env | `Init env ->
              let ps, message = env.data in
              One_or_two.map ps ~f:(fun p -> (p, message)) |> One_or_two.to_list )
        in
        let open Deferred.Or_error.Let_syntax in
        let%map result = Verifier.verify_transaction_snarks verifier ps in
        match result with
        | Ok () ->
            List.map ps0 ~f:(fun _ -> `Valid ())
        | Error err ->
            List.map ps0 ~f:(function `Partially_validated env | `Init env ->
                `Potentially_invalid (env, err) ) )

  module Work_key = struct
    module T = struct
      type t =
        (Transaction_snark.Statement.t One_or_two.t * Mina_base.Sok_message.t)
        Envelope.Incoming.t
      [@@deriving sexp, compare]
    end

    let of_proof_envelope t =
      Envelope.Incoming.map t ~f:(fun (ps, message) ->
          (One_or_two.map ~f:Ledger_proof.statement ps, message) )

    include T
    include Comparable.Make (T)
  end

  let verify' (t : t) ps =
    let open Deferred.Or_error.Let_syntax in
    let%map invalid =
      Deferred.Or_error.List.filter_map ps ~f:(fun p ->
          match%map verify t p with true -> None | false -> Some p )
    in
    `Invalid
      (Work_key.Set.of_list (List.map invalid ~f:Work_key.of_proof_envelope))

  let%test_module "With valid and invalid proofs" =
    ( module struct
      open Mina_base

      let precomputed_values = Lazy.force Precomputed_values.for_unit_tests

      let proof_level = precomputed_values.proof_level

      let constraint_constants = precomputed_values.constraint_constants

      let logger = Logger.null ()

      let verifier =
        Async.Thread_safe.block_on_async_exn (fun () ->
            Verifier.create ~logger ~proof_level ~constraint_constants
              ~conf_dir:None
              ~pids:(Child_processes.Termination.create_pid_table ())
              ~commit_id:"not specified for unit tests" () )

      let gen_proofs =
        let open Quickcheck.Generator.Let_syntax in
        let data_gen =
          let%bind statements =
            One_or_two.gen Transaction_snark.Statement.gen
          in
          let%map { fee; prover } = Fee_with_prover.gen in
          let message = Mina_base.Sok_message.create ~fee ~prover in
          ( One_or_two.map statements ~f:Ledger_proof.For_tests.mk_dummy_proof
          , message )
        in
        Envelope.Incoming.gen data_gen

      let gen_invalid_proofs =
        let open Quickcheck.Generator.Let_syntax in
        let data_gen =
          let%bind statements =
            One_or_two.gen Transaction_snark.Statement.gen
          in
          let%bind { fee; prover } = Fee_with_prover.gen in
          let%map invalid_prover =
            Quickcheck.Generator.filter Signature_lib.Public_key.Compressed.gen
              ~f:(Signature_lib.Public_key.Compressed.( <> ) prover)
          in
          let sok_digest =
            Mina_base.Sok_message.(digest (create ~fee ~prover:invalid_prover))
          in
          let message = Mina_base.Sok_message.create ~fee ~prover in
          ( One_or_two.map statements ~f:(fun statement ->
                Ledger_proof.create ~statement ~sok_digest
                  ~proof:(Lazy.force Proof.transaction_dummy) )
          , message )
        in
        Envelope.Incoming.gen data_gen

      let run_test proof_lists =
        let batcher = create verifier in
        Deferred.List.iter proof_lists ~f:(fun (invalid_proofs, proof_list) ->
            let%map r = verify' batcher proof_list in
            let (`Invalid ps) = Or_error.ok_exn r in
            assert (Work_key.Set.equal ps invalid_proofs) )

      let gen ~(valid_count : [ `Any | `Count of int ])
          ~(invalid_count : [ `Any | `Count of int ]) =
        let open Quickcheck.Generator.Let_syntax in
        let gen_with_count count gen =
          match count with
          | `Any ->
              Quickcheck.Generator.list_non_empty gen
          | `Count c ->
              Quickcheck.Generator.list_with_length c gen
        in
        let invalid_gen = gen_with_count invalid_count gen_invalid_proofs in
        let valid_gen = gen_with_count valid_count gen_proofs in
        let%map lst =
          Quickcheck.Generator.(list (both valid_gen invalid_gen))
        in
        List.map lst ~f:(fun (valid, invalid) ->
            ( Work_key.(Set.of_list (List.map ~f:of_proof_envelope invalid))
            , List.permute valid @ invalid ) )

      let%test_unit "all valid proofs" =
        Quickcheck.test ~trials:10
          (gen ~valid_count:`Any ~invalid_count:(`Count 0))
          ~f:(fun proof_lists ->
            Async.Thread_safe.block_on_async_exn (fun () ->
                run_test proof_lists ) )

      let%test_unit "some invalid proofs" =
        Quickcheck.test ~trials:10
          (gen ~valid_count:`Any ~invalid_count:`Any)
          ~f:(fun proof_lists ->
            Async.Thread_safe.block_on_async_exn (fun () ->
                run_test proof_lists ) )

      let%test_unit "all invalid proofs" =
        Quickcheck.test ~trials:10
          (gen ~valid_count:(`Count 0) ~invalid_count:`Any)
          ~f:(fun proof_lists ->
            Async.Thread_safe.block_on_async_exn (fun () ->
                run_test proof_lists ) )
    end )
end

1	open Core_kernel	1✔
2	open Async_kernel
3	open Network_peer
4
5	(* Only show stdout for failed inline tests. *)
6	open Inline_test_quiet_logs
7
8	module Id = Unique_id.Int ()
9
10	type ('init, 'result) elt =	×
11	{ id : Id.t	×
12	; data : 'init	×
13	; weight : int	×
14	; res : (('result, Verifier.invalid) Result.t Or_error.t Ivar.t[@sexp.opaque])	×
15	}
16	[@@deriving sexp]
17
18	type ('proof, 'result) state =	×
19	\| Waiting	×
20	\| Verifying of { out_for_verification : ('proof, 'result) elt list }	×
21	[@@deriving sexp]
22
23	module Q = Doubly_linked
24
25	type ('init, 'partially_validated, 'result) t =	×
26	{ mutable state : ('init, 'result) state	×
27	; how_to_add : [ `Insert \| `Enqueue_back ]	×
28	; queue : ('init, 'result) elt Q.t	×
29	; compare_init : ('init -> 'init -> int) option	×
30	; logger : (Logger.t[@sexp.opaque])	×
31	; weight : 'init -> int	×
32	; max_weight_per_call : int option	×
33	; verifier :	×
34	( (* The batched verifier may make partial progress on its input so that we can
35	save time when it is re-verified in a smaller batch in the case that a batch
36	fails to verify.
37	*)
38	[ `Init of 'init \| `Partially_validated of 'partially_validated ] list
39	-> [ `Valid of 'result
40	\| Verifier.invalid
41	\| `Potentially_invalid of 'partially_validated * Error.t ]
42	list
43	Deferred.Or_error.t
44	[@sexp.opaque] )
45	}
46	[@@deriving sexp]
47
48	let create ?(how_to_add = `Enqueue_back) ?logger ?compare_init	×
49	?(weight = fun _ -> 1) ?max_weight_per_call verifier =	×
50	{ state = Waiting	×
51	; queue = Q.create ()	×
52	; how_to_add
53	; compare_init
54	; verifier
55	; weight
56	; max_weight_per_call
57	; logger = Option.value logger ~default:(Logger.create ())	×
58	}
59
60	let call_verifier t (ps : 'proof list) = t.verifier ps	×
61
62	(* Worst case (if all the proofs are invalid): log n * (2^(log n) + 1)
63	In the average case this should show better performance.
64	We could implement the trusted/untrusted batches from the snark pool batching RFC #4882
65	to further mitigate possible DoS/DDoS here
66	*)
67	let rec determine_outcome :
68	type p r partial.
69	(p, r) elt list
70	-> [ `Valid of r
71	\| `Potentially_invalid of partial * Error.t
72	\| Verifier.invalid ]
73	list
74	-> (p, partial, r) t
75	-> unit Deferred.Or_error.t =
76	fun ps res v ->
77	O1trace.thread "determining_batcher_outcome" (fun () ->	×
78	(* First separate out all the known results. That information will definitely be included
79	in the outcome.
80	*)
81	let logger = Logger.create () in	×
82	let potentially_invalid =	×
83	List.filter_map (List.zip_exn ps res) ~f:(fun (elt, r) ->	×
84	match r with	×
85	\| `Valid r ->	×
86	if Ivar.is_full elt.res then
87	[%log error] "Ivar.fill bug is here!" ;	×
88	Ivar.fill elt.res (Ok (Ok r)) ;	×
89	None	×
90	\| `Invalid_keys keys ->	×
91	if Ivar.is_full elt.res then
92	[%log error] "Ivar.fill bug is here!" ;	×
93	Ivar.fill elt.res (Ok (Error (`Invalid_keys keys))) ;	×
94	None	×
95	\| `Invalid_signature keys ->	×
96	if Ivar.is_full elt.res then
97	[%log error] "Ivar.fill bug is here!" ;	×
98	Ivar.fill elt.res (Ok (Error (`Invalid_signature keys))) ;	×
99	None	×
100	\| `Invalid_proof err ->	×
101	if Ivar.is_full elt.res then
102	[%log error] "Ivar.fill bug is here!" ;	×
103	Ivar.fill elt.res (Ok (Error (`Invalid_proof err))) ;	×
104	None	×
105	\| `Missing_verification_key keys ->	×
106	if Ivar.is_full elt.res then
107	[%log error] "Ivar.fill bug is here!" ;	×
108	Ivar.fill elt.res (Ok (Error (`Missing_verification_key keys))) ;	×
109	None	×
110	\| `Unexpected_verification_key keys ->	×
111	if Ivar.is_full elt.res then
112	[%log error] "Ivar.fill bug is here!" ;	×
113	Ivar.fill elt.res	×
114	(Ok (Error (`Unexpected_verification_key keys))) ;
115	None	×
116	\| `Mismatched_authorization_kind keys ->	×
117	if Ivar.is_full elt.res then
118	[%log error] "Ivar.fill bug is here!" ;	×
119	Ivar.fill elt.res	×
120	(Ok (Error (`Mismatched_authorization_kind keys))) ;
121	None	×
122	\| `Potentially_invalid (new_hint, err) ->	×
123	Some (elt, new_hint, err) )
124	in
125	let open Deferred.Or_error.Let_syntax in	×
126	match potentially_invalid with
127	\| [] ->	×
128	(* All results are known *)
129	return ()
130	\| [ ({ res; _ }, _, err) ] ->	×
131	if Ivar.is_full res then [%log error] "Ivar.fill bug is here!" ;	×
132	Ivar.fill res (Ok (Error (`Invalid_proof err))) ;	×
133	(* If there is a potentially invalid proof in this batch of size 1, then
134	that proof is itself invalid. *)
135	return ()	×
136	\| _ ->	×
137	let outcome xs =
138	let%bind res_xs =
139	call_verifier v	×
140	(List.map xs ~f:(fun (_e, new_hint, _) ->	×
141	`Partially_validated new_hint ) )	×
142	in
143	determine_outcome (List.map xs ~f:(fun (e, _, _) -> e)) res_xs v	×
144	in
145	let length = List.length potentially_invalid in
146	let left, right = List.split_n potentially_invalid (length / 2) in	×
147	let%bind () = outcome left in	×
148	outcome right )	×
149
150	let compare_elt ~compare t1 t2 =
151	match compare t1.data t2.data with 0 -> Id.compare t1.id t2.id \| x -> x	×
152
153	let order_proofs t =
154	match t.compare_init with	×
155	\| None ->	×
156	Fn.id
157	\| Some compare ->	×
158	List.sort ~compare:(compare_elt ~compare)
159
160	(* When new proofs come in put them in the queue.
161	If state = Waiting, verify those proofs immediately.
162	Whenever the verifier returns, if the queue is nonempty, flush it into the verifier.
163	*)
164
165	let rec start_verifier : type proof partial r. (proof, partial, r) t -> unit =
166	fun t ->
167	O1trace.sync_thread "running_batcher_verifier_loop" (fun () ->	×
168	if Q.is_empty t.queue then	×
169	(* we looped in the else after verifier finished but no pending work. *)
170	t.state <- Waiting	×
171	else (	×
172	[%log' debug t.logger] "Verifying proofs in batch of size $num_proofs"	×
173	~metadata:[ ("num_proofs", `Int (Q.length t.queue)) ] ;	×
174	let out_for_verification =	×
175	let proofs =
176	match t.max_weight_per_call with
177	\| None ->	×
178	let proofs = Q.to_list t.queue in
179	Q.clear t.queue ; proofs	×
180	\| Some max_weight ->	×
181	let rec take capacity acc =
182	match Q.first t.queue with	×
183	\| None ->	×
184	acc
185	\| Some ({ weight; _ } as proof) ->	×
186	if weight <= capacity then (	×
187	ignore (Q.remove_first t.queue : (proof, r) elt option) ;	×
188	take (capacity - weight) (proof :: acc) )
189	else acc	×
190	in
191	List.rev (take max_weight [])	×
192	in
193	order_proofs t proofs	×
194	in
195	[%log' debug t.logger] "Calling verifier with $num_proofs on $ids"	×
196	~metadata:
197	[ ("num_proofs", `Int (List.length out_for_verification))	×
198	; ( "ids"
199	, `List
200	(List.map	×
201	~f:(fun { id; _ } -> `Int (Id.to_int_exn id))	×
202	out_for_verification ) )
203	] ;
204	let res =	×
205	match%bind
206	call_verifier t	×
207	(List.map out_for_verification ~f:(fun { data = p; _ } -> `Init p))	×
208	with
209	\| Error e ->	×
210	Deferred.return (Error e)
211	\| Ok res ->	×
212	determine_outcome out_for_verification res t
213	in
214	t.state <- Verifying { out_for_verification } ;	×
215	upon res (fun r ->
216	( match r with	×
217	\| Ok () ->	×
218	()
219	\| Error e ->	×
220	List.iter out_for_verification ~f:(fun x ->	×
221	Ivar.fill_if_empty x.res (Error e) ) ) ;	×
222	start_verifier t ) ) )
223
224	let verify (type p r partial) (t : (p, partial, r) t) (proof : p) :
225	(r, Verifier.invalid) Result.t Deferred.Or_error.t =
226	let elt =	×
227	{ id = Id.create ()	×
228	; data = proof
229	; weight = t.weight proof	×
230	; res = Ivar.create ()	×
231	}
232	in
233	ignore
234	( match (t.how_to_add, t.compare_init) with
235	\| `Enqueue_back, _ \| `Insert, None ->	×
236	Q.insert_last t.queue elt	×
237	\| `Insert, Some compare -> (	×
238	(* Find the first element that [proof] is less than *)
239	let compare = compare_elt ~compare in
240	match Q.find_elt t.queue ~f:(fun e -> compare elt e < 0) with	×
241	\| None ->	×
242	(* [proof] is greater than all elts in the queue, and so goes in the back. *)
243	Q.insert_last t.queue elt	×
244	\| Some succ ->	×
245	Q.insert_before t.queue succ elt )	×
246	: (p, r) elt Q.Elt.t ) ;
247	(match t.state with Verifying _ -> () \| Waiting -> start_verifier t) ;	×
248	Ivar.read elt.res
249
250	type ('a, 'b, 'c) batcher = ('a, 'b, 'c) t [@@deriving sexp]	×
251
252	let compare_envelope (e1 : _ Envelope.Incoming.t) (e2 : _ Envelope.Incoming.t) =
253	Envelope.Sender.compare e1.sender e2.sender	×
254
255	module Transaction_pool = struct
256	open Mina_base
257
258	type diff = User_command.Verifiable.t list Envelope.Incoming.t	×
259	[@@deriving sexp]
260
261	(* A partially verified transaction is either valid, or valid assuming that some list of
262	(verification key, statement, proof) triples will verify. That is, the transaction has
263	already been validated in all ways, except the proofs were in a batch that failed to
264	verify.
265	*)
266	type partial_item =	×
267	[ `Valid of User_command.Valid.t	×
268	\| `Valid_assuming of
269	User_command.Verifiable.t	×
270	* ( Pickles.Side_loaded.Verification_key.t	×
271	* Zkapp_statement.t
272	* Pickles.Side_loaded.Proof.t )
273	list ]	×
274	[@@deriving sexp]	×
275
276	type partial = partial_item list [@@deriving sexp]	×
277
278	type t = (diff, partial, User_command.Valid.t list) batcher [@@deriving sexp]	×
279
280	type input = [ `Init of diff \| `Partially_validated of partial ]
281
282	let init_result (ds : input list) =
283	(* We store a result for every diff in the input. *)
284	Array.of_list_map ds ~f:(function	×
285	\| `Init d ->	×
286	(* Initially, the status of all the transactions in a never-before-seen
287	diff are unknown. *)
288	`In_progress (Array.of_list_map d.data ~f:(fun _ -> `Unknown))	×
289	\| `Partially_validated d ->	×
290	(* We've seen this diff before, so we have some information about its
291	transactions. *)
292	`In_progress
293	(Array.of_list_map d ~f:(function	×
294	\| `Valid c ->	×
295	`Valid c
296	\| `Valid_assuming x ->	×
297	`Valid_assuming x ) ) )
298
299	let list_of_array_map a ~f = List.init (Array.length a) ~f:(fun i -> f a.(i))	×
300
301	let all_valid a =
302	Option.all	×
303	(Array.to_list	×
304	(Array.map a ~f:(function `Valid c -> Some c \| _ -> None)) )	×
305
306	let create verifier : t =
307	let logger = Logger.create () in	×
308	create ~compare_init:compare_envelope ~logger (fun (ds : input list) ->	×
309	O1trace.thread "dispatching_transaction_pool_batcher_verification"	×
310	(fun () ->
311	[%log debug]	×
312	"Dispatching $num_proofs transaction pool proofs to verifier"
313	~metadata:[ ("num_proofs", `Int (List.length ds)) ] ;	×
314	let open Deferred.Or_error.Let_syntax in	×
315	let result = init_result ds in
316	(* Extract all the transactions that have not yet been fully validated and hold on to their
317	position (diff index, position in diff). *)
318	let unknowns =	×
319	List.concat_mapi ds ~f:(fun i x ->
320	match x with	×
321	\| `Init diff ->	×
322	List.mapi diff.data ~f:(fun j c -> ((i, j), c))	×
323	\| `Partially_validated partial ->	×
324	List.filter_mapi partial ~f:(fun j c ->
325	match c with	×
326	\| `Valid _ ->	×
327	None
328	\| `Valid_assuming (v, _) ->	×
329	(* TODO: This rechecks the signatures on zkApp transactions... oh well for now *)
330	Some ((i, j), v) ) )
331	in
332	let%map res =
333	(* Verify the unknowns *)
334	Verifier.verify_commands verifier	×
335	(List.map unknowns ~f:(fun (_, txn) ->	×
336	{ With_status.data = txn; status = Applied } ) )	×
337	in
338	(* We now iterate over the results of the unknown transactions and appropriately modify
339	the verification result of the diff that it belongs to. *)
340	List.iter2_exn unknowns res ~f:(fun ((i, j), v) r ->	×
341	match r with	×
342	\| `Invalid_keys keys ->	×
343	(* A diff is invalid is any of the transactions it contains are invalid.
344	Invalidate the whole diff that this transaction comes from. *)
345	result.(i) <- `Invalid_keys keys
346	\| `Invalid_signature keys ->	×
347	(* Invalidate the whole diff *)
348	result.(i) <- `Invalid_signature keys
349	\| `Missing_verification_key keys ->	×
350	(* Invalidate the whole diff *)
351	result.(i) <- `Missing_verification_key keys
352	\| `Unexpected_verification_key keys ->	×
353	(* Invalidate the whole diff *)
354	result.(i) <- `Unexpected_verification_key keys
355	\| `Mismatched_authorization_kind keys ->	×
356	(* Invalidate the whole diff *)
357	result.(i) <- `Mismatched_authorization_kind keys
358	\| `Invalid_proof err ->	×
359	(* Invalidate the whole diff *)
360	result.(i) <- `Invalid_proof err
361	\| `Valid_assuming xs -> (	×
362	match result.(i) with
363	\| `Invalid_keys _	×
364	\| `Invalid_signature _	×
365	\| `Invalid_proof _	×
366	\| `Missing_verification_key _	×
367	\| `Unexpected_verification_key _	×
368	\| `Mismatched_authorization_kind _ ->	×
369	(* If this diff has already been declared invalid, knowing that one of its
370	transactions is partially valid is not useful. *)
371	()
372	\| `In_progress a ->	×
373	(* The diff may still be valid. *)
374	a.(j) <- `Valid_assuming (v, xs) )
375	\| `Valid c -> (	×
376	(* Similar to the above. *)
377	match result.(i) with
378	\| `Invalid_keys _	×
379	\| `Invalid_signature _	×
380	\| `Invalid_proof _	×
381	\| `Missing_verification_key _	×
382	\| `Unexpected_verification_key _	×
383	\| `Mismatched_authorization_kind _ ->	×
384	()
385	\| `In_progress a ->	×
386	a.(j) <- `Valid c ) ) ;
387	list_of_array_map result ~f:(function	×
388	\| `Invalid_keys keys ->	×
389	`Invalid_keys keys
390	\| `Invalid_signature keys ->	×
391	`Invalid_signature keys
392	\| `Invalid_proof err ->	×
393	`Invalid_proof err
394	\| `Missing_verification_key keys ->	×
395	`Missing_verification_key keys
396	\| `Unexpected_verification_key keys ->	×
397	`Unexpected_verification_key keys
398	\| `Mismatched_authorization_kind keys ->	×
399	`Mismatched_authorization_kind keys
400	\| `In_progress a -> (	×
401	(* If the diff is all valid, we're done. If not, we return a partial
402	result. *)
403	match all_valid a with
404	\| Some res ->	×
405	`Valid res
406	\| None ->	×
407	`Potentially_invalid
408	( list_of_array_map a ~f:(function	×
409	\| `Unknown ->
410	assert false
411	\| `Valid c ->	×
412	`Valid c
413	\| `Valid_assuming (v, xs) ->	×
414	`Valid_assuming (v, xs) )
415	, Error.of_string "In progress" ) ) ) ) )	×
416
417	let verify (t : t) = verify t	×
418	end
419
420	module Snark_pool = struct
421	type proof_envelope =	×
422	(Ledger_proof.t One_or_two.t * Mina_base.Sok_message.t) Envelope.Incoming.t	×
423	[@@deriving sexp]
424
425	(* We don't use partial verification here. *)
426	type partial = proof_envelope [@@deriving sexp]	×
427
428	type t = (proof_envelope, partial, unit) batcher [@@deriving sexp]	×
429
430	let verify (t : t) (p : proof_envelope) : bool Deferred.Or_error.t =
431	let open Deferred.Or_error.Let_syntax in	×
432	match%map verify t p with Ok () -> true \| Error _ -> false	×
433
434	let create verifier : t =
435	let logger = Logger.create () in	×
436	create	×
437	(* TODO: Make this a proper config detail once we have data on what a
438	good default would be.
439	*)
440	~max_weight_per_call:
441	(Option.value_map ~default:1000 ~f:Int.of_string	×
442	(Sys.getenv_opt "MAX_VERIFIER_BATCH_SIZE") )	×
443	~compare_init:compare_envelope ~logger
444	(fun ps0 ->
445	[%log debug] "Dispatching $num_proofs snark pool proofs to verifier"	×
446	~metadata:[ ("num_proofs", `Int (List.length ps0)) ] ;	×
447	let ps =	×
448	List.concat_map ps0 ~f:(function
449	\| `Partially_validated env \| `Init env ->	×
450	let ps, message = env.data in
451	One_or_two.map ps ~f:(fun p -> (p, message)) \|> One_or_two.to_list )	×
452	in
453	let open Deferred.Or_error.Let_syntax in	×
454	let%map result = Verifier.verify_transaction_snarks verifier ps in	×
455	match result with	×
456	\| Ok () ->	×
457	List.map ps0 ~f:(fun _ -> `Valid ())	×
458	\| Error err ->	×
459	List.map ps0 ~f:(function `Partially_validated env \| `Init env ->	×
460	`Potentially_invalid (env, err) ) )
461
462	module Work_key = struct
463	module T = struct
464	type t =	×
465	(Transaction_snark.Statement.t One_or_two.t * Mina_base.Sok_message.t)	×
466	Envelope.Incoming.t	×
467	[@@deriving sexp, compare]
468	end
469
470	let of_proof_envelope t =
471	Envelope.Incoming.map t ~f:(fun (ps, message) ->	×
472	(One_or_two.map ~f:Ledger_proof.statement ps, message) )	×
473
474	include T
475	include Comparable.Make (T)
476	end
477
478	let verify' (t : t) ps =
479	let open Deferred.Or_error.Let_syntax in	×
480	let%map invalid =
481	Deferred.Or_error.List.filter_map ps ~f:(fun p ->	×
482	match%map verify t p with true -> None \| false -> Some p )	×
483	in
484	`Invalid	×
485	(Work_key.Set.of_list (List.map invalid ~f:Work_key.of_proof_envelope))	×
486
487	let%test_module "With valid and invalid proofs" =
488	( module struct
489	open Mina_base
490
491	let precomputed_values = Lazy.force Precomputed_values.for_unit_tests	×
492
493	let proof_level = precomputed_values.proof_level
494
495	let constraint_constants = precomputed_values.constraint_constants
496
497	let logger = Logger.null ()	×
498
499	let verifier =
500	Async.Thread_safe.block_on_async_exn (fun () ->	×
501	Verifier.create ~logger ~proof_level ~constraint_constants	×
502	~conf_dir:None
503	~pids:(Child_processes.Termination.create_pid_table ())	×
504	~commit_id:"not specified for unit tests" () )
505
506	let gen_proofs =
507	let open Quickcheck.Generator.Let_syntax in
508	let data_gen =
509	let%bind statements =
510	One_or_two.gen Transaction_snark.Statement.gen	×
511	in
512	let%map { fee; prover } = Fee_with_prover.gen in
513	let message = Mina_base.Sok_message.create ~fee ~prover in	×
514	( One_or_two.map statements ~f:Ledger_proof.For_tests.mk_dummy_proof	×
515	, message )
516	in
517	Envelope.Incoming.gen data_gen	×
518
519	let gen_invalid_proofs =
520	let open Quickcheck.Generator.Let_syntax in
521	let data_gen =
522	let%bind statements =
523	One_or_two.gen Transaction_snark.Statement.gen	×
524	in
525	let%bind { fee; prover } = Fee_with_prover.gen in
526	let%map invalid_prover =
527	Quickcheck.Generator.filter Signature_lib.Public_key.Compressed.gen	×
528	~f:(Signature_lib.Public_key.Compressed.( <> ) prover)	×
529	in
530	let sok_digest =	×
531	Mina_base.Sok_message.(digest (create ~fee ~prover:invalid_prover))	×
532	in
533	let message = Mina_base.Sok_message.create ~fee ~prover in
534	( One_or_two.map statements ~f:(fun statement ->	×
535	Ledger_proof.create ~statement ~sok_digest	×
536	~proof:(Lazy.force Proof.transaction_dummy) )	×
537	, message )
538	in
539	Envelope.Incoming.gen data_gen	×
540
541	let run_test proof_lists =
542	let batcher = create verifier in	×
543	Deferred.List.iter proof_lists ~f:(fun (invalid_proofs, proof_list) ->	×
544	let%map r = verify' batcher proof_list in	×
545	let (`Invalid ps) = Or_error.ok_exn r in	×
546	assert (Work_key.Set.equal ps invalid_proofs) )	×
547
548	let gen ~(valid_count : [ `Any \| `Count of int ])
549	~(invalid_count : [ `Any \| `Count of int ]) =
550	let open Quickcheck.Generator.Let_syntax in	×
551	let gen_with_count count gen =
552	match count with	×
553	\| `Any ->	×
554	Quickcheck.Generator.list_non_empty gen
555	\| `Count c ->	×
556	Quickcheck.Generator.list_with_length c gen
557	in
558	let invalid_gen = gen_with_count invalid_count gen_invalid_proofs in
559	let valid_gen = gen_with_count valid_count gen_proofs in	×
560	let%map lst =
561	Quickcheck.Generator.(list (both valid_gen invalid_gen))	×
562	in
563	List.map lst ~f:(fun (valid, invalid) ->	×
564	( Work_key.(Set.of_list (List.map ~f:of_proof_envelope invalid))	×
565	, List.permute valid @ invalid ) )	×
566
567	let%test_unit "all valid proofs" =
568	Quickcheck.test ~trials:10	×
569	(gen ~valid_count:`Any ~invalid_count:(`Count 0))
570	~f:(fun proof_lists ->
571	Async.Thread_safe.block_on_async_exn (fun () ->	×
572	run_test proof_lists ) )	×
573
574	let%test_unit "some invalid proofs" =
575	Quickcheck.test ~trials:10	×
576	(gen ~valid_count:`Any ~invalid_count:`Any)
577	~f:(fun proof_lists ->
578	Async.Thread_safe.block_on_async_exn (fun () ->	×
579	run_test proof_lists ) )	×
580
581	let%test_unit "all invalid proofs" =
582	Quickcheck.test ~trials:10	×
583	(gen ~valid_count:(`Count 0) ~invalid_count:`Any)
584	~f:(fun proof_lists ->
585	Async.Thread_safe.block_on_async_exn (fun () ->	×
586	run_test proof_lists ) )	×
587	end )
588	end	1✔

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