Merge branch 'lyh/update-snark-worker-rpc-jun-12' into lyh/update-snark-worker-rpc-jun-12-testing

Author: glyh

src/lib/mina_lib/mina_lib.ml

@@ -914,20 +914,41 @@ let add_work t (work : Snark_work_lib.Result.Partitioned.Stable.Latest.t) =
       `Removed
   | Processed None ->
       `Ok
-  | Processed (Some (stmts, priced_proof)) ->
+  | Processed (Some { spec_with_proof; fee; prover }) ->
+      let proofs = spec_with_proof |> One_or_two.map ~f:Tuple2.get2 in
+      let fee_with_prover = Fee_with_prover.{ fee; prover } in
+      let spec = One_or_two.map ~f:Tuple2.get1 spec_with_proof in
+      let stmts =
+        spec_with_proof
+        |> One_or_two.map ~f:(fun (spec, _) ->
+               Snark_work_lib.Selector.Single.Spec.Poly.statement spec )
+      in
       [%log' info t.config.logger] "Partitioner combined work"
         ~metadata:
           [ ( "stmts"
             , One_or_two.to_yojson Mina_state.Snarked_ledger_state.to_yojson
                 stmts )
-          ; ("fee_with_prover", Fee_with_prover.to_yojson priced_proof.fee)
-          ; ( "proofs"
-            , One_or_two.to_yojson Ledger_proof.to_yojson priced_proof.proof )
+          ; ("fee_with_prover", Fee_with_prover.to_yojson fee_with_prover)
+          ; ("proofs", proofs |> One_or_two.to_yojson Ledger_proof.to_yojson)
           ] ;
       ignore (Or_error.try_with (fun () -> update_metrics ()) : unit Or_error.t) ;
       Network_pool.Snark_pool.(
         Local_sink.push t.pipes.snark_local_sink
-          (Add_solved_work (stmts, priced_proof), ignore))
+          ( Add_solved_work
+              ( stmts
+              , Network_pool.Priced_proof.
+                  { proof = proofs; fee = fee_with_prover } )
+          , Result.iter_error ~f:(fun err ->
+                [%log' info t.config.logger]
+                  "Failed to push completed work to local snark sink, \
+                   returning them to work selector"
+                  ~metadata:
+                    [ ( "stmts"
+                      , One_or_two.to_yojson
+                          Mina_state.Snarked_ledger_state.to_yojson stmts )
+                    ; ("error", `String (Error.to_string_hum err))
+                    ] ;
+                Work_selector.add_back t.work_selector spec ) ))
       |> Deferred.don't_wait_for ;
       `Ok
 

src/lib/snark_work_lib/combined_result.mli

@@ -1,5 +1,5 @@
-open Mina_wire_types
-
 type t =
-  Transaction_snark_work.Statement.V2.t
-  * Ledger_proof.V2.t One_or_two.V1.t Network_pool_priced_proof.V1.t
+  { spec_with_proof : (Single_spec.t * Ledger_proof.t) One_or_two.t
+  ; fee : Currency.Fee.t
+  ; prover : Signature_lib.Public_key.Compressed.t
+  }

src/lib/work_partitioner/combining_result.ml

@@ -31,42 +31,32 @@ let finalize_one ~submitted_result ~spec ~fee ~prover =
       ~f_proof:Ledger_proof.Cached.read_proof_from_disk submitted_result
   in
   let Snark_work_lib.Result.Single.Poly.{ proof; _ } = submitted_result in
-  let statements =
-    `One (Snark_work_lib.Selector.Single.Spec.Poly.statement spec)
-  in
-  Done
-    ( statements
-    , Mina_wire_types.Network_pool_priced_proof.V1.
-        { proof = `One proof; fee = { fee; prover } } )
+  Done { spec_with_proof = `One (spec, proof); fee; prover }
 
 let finalize_two ~submitted_result ~other_spec ~in_pool_result ~submitted_half
     ~fee ~prover =
   let submitted_result =
-    Snark_work_lib.Result.Single.Poly.map ~f_spec:(const other_spec)
-      ~f_proof:Ledger_proof.Cached.read_proof_from_disk submitted_result
+    let Snark_work_lib.Result.Single.Poly.{ spec; proof; _ } =
+      Snark_work_lib.Result.Single.Poly.map ~f_spec:(const other_spec)
+        ~f_proof:Ledger_proof.Cached.read_proof_from_disk submitted_result
+    in
+    (spec, proof)
   in
   let in_pool_result =
-    Snark_work_lib.Result.Single.Poly.map ~f_spec:Fn.id
-      ~f_proof:Ledger_proof.Cached.read_proof_from_disk in_pool_result
+    let Snark_work_lib.Result.Single.Poly.{ spec; proof; _ } =
+      Snark_work_lib.Result.Single.Poly.map ~f_spec:Fn.id
+        ~f_proof:Ledger_proof.Cached.read_proof_from_disk in_pool_result
+    in
+    (spec, proof)
   in
-  let results =
+  let spec_with_proof =
     match submitted_half with
     | `First ->
         `Two (submitted_result, in_pool_result)
     | `Second ->
         `Two (in_pool_result, submitted_result)
   in
-  let statements =
-    One_or_two.map
-      ~f:(fun result ->
-        Snark_work_lib.Selector.Single.Spec.Poly.statement result.spec )
-      results
-  in
-  let proof = One_or_two.map ~f:(fun result -> result.proof) results in
-  Done
-    ( statements
-    , Mina_wire_types.Network_pool_priced_proof.V1.
-        { proof; fee = { fee; prover } } )
+  Done { spec_with_proof; fee; prover }
 
 let merge_single_result (current : t) ~logger
     ~(submitted_result :

src/lib/work_selector/intf.ml

@@ -136,6 +136,16 @@ module type Lib_intf = sig
          Snark_work_lib.Work.Single.Spec.t
          One_or_two.t
          list
+
+    (** [add_back t work] add back a work manually, usually this is needed when 
+        proof verification failed *)
+    val add_back :
+         t
+      -> ( Transaction_witness.t
+         , Ledger_proof.Cached.t )
+         Snark_work_lib.Work.Single.Spec.t
+         One_or_two.t
+      -> unit
   end
 
   (**jobs that are not in the snark pool yet*)

src/lib/work_selector/work_lib.ml

@@ -103,11 +103,18 @@ module Make (Inputs : Intf.Inputs_intf) = struct
           Fee.compare fee competing_fee < 0 )
 
     let all_unscheduled_expensive_works ~snark_pool ~fee (t : t) =
+      (* WARN: Always use filtering as [add_back] may causes work pool to have duplicates *)
       O1trace.sync_thread "work_lib_all_unscheduled_expensive_works" (fun () ->
           List.filter t.available_jobs ~f:(fun job ->
               let job_key = Job_key.of_job job in
               (not (Job_key_set.mem t.jobs_scheduled job_key))
               && does_not_have_better_fee ~snark_pool ~fee job_key ) )
+
+    let add_back (state : t) x =
+      state.jobs_scheduled <-
+        Job_key_set.remove state.jobs_scheduled
+          (One_or_two.map ~f:Work_spec.statement x) ;
+      state.available_jobs <- x :: state.available_jobs
   end
 
   let all_pending_work ~snark_pool statements =

src/lib/work_selector/work_selector.ml

@@ -26,3 +26,5 @@ let pending_work_statements = Lib.pending_work_statements
 let all_work = Lib.all_work
 
 let completed_work_statements = Lib.completed_work_statements
+
+let add_back = Lib.State.add_back

src/lib/work_selector/work_selector.mli

@@ -39,3 +39,5 @@ val all_work :
 
 val completed_work_statements :
   snark_pool:snark_pool -> State.t -> Transaction_snark_work.Checked.t list
+
+val add_back : State.t -> work One_or_two.t -> unit