Merge pull request #124 from zeko-labs/MartinOndejka/commands-queue

Implement commands queue execution

src/app/zeko/sequencer/lib/da_layer.ml

@@ -11,10 +11,7 @@ let command_to_yojson command =
   | User_command.Signed_command signed_command ->
       `Assoc
         [ ("commandType", `Int 0)
-        ; ( "data"
-          , `String
-              (Signed_command.to_base64
-                 (Signed_command.forget_check signed_command) ) )
+        ; ("data", `String (Signed_command.to_base64 signed_command))
         ]
   | User_command.Zkapp_command zkapp_command ->
       `Assoc

src/app/zeko/sequencer/lib/executor.ml

@@ -0,0 +1,88 @@
+open Async
+open Async_kernel
+open Core_kernel
+open Mina_base
+open Signature_lib
+
+type t =
+  { l1_uri : Uri.t Cli_lib.Flag.Types.with_name
+  ; signer : Keypair.t
+  ; q : unit Throttle.t
+  ; mutable nonce : int option
+  ; max_attempts : int
+  ; delay : Time.Span.t
+  }
+
+let create ?(max_attempts = 5) ?(delay = Time.Span.of_sec 5.) ~l1_uri ~signer ()
+    =
+  { l1_uri
+  ; signer
+  ; q = Throttle.create ~continue_on_error:false ~max_concurrent_jobs:1
+  ; nonce = None
+  ; max_attempts
+  ; delay
+  }
+
+let refresh_nonce t = t.nonce <- None
+
+let process_command t (command : Zkapp_command.t) =
+  let rec retry attempt () =
+    let%bind nonce =
+      match t.nonce with
+      | Some nonce ->
+          return nonce
+      | None ->
+          Gql_client.fetch_nonce t.l1_uri
+            (Public_key.compress t.signer.public_key)
+    in
+    let command =
+      { command with
+        fee_payer =
+          { command.fee_payer with
+            body =
+              { command.fee_payer.body with
+                nonce = Unsigned.UInt32.of_int nonce
+              }
+          }
+      }
+    in
+    let full_commitment =
+      Zkapp_command.Transaction_commitment.create_complete
+        (Zkapp_command.commitment command)
+        ~memo_hash:(Signed_command_memo.hash command.memo)
+        ~fee_payer_hash:
+          (Zkapp_command.Digest.Account_update.create
+             (Account_update.of_fee_payer command.fee_payer) )
+    in
+    let signature =
+      Signature_lib.Schnorr.Chunked.sign
+        ~signature_kind:Mina_signature_kind.Testnet t.signer.private_key
+        (Random_oracle.Input.Chunked.field full_commitment)
+    in
+    let command =
+      { command with
+        fee_payer = { command.fee_payer with authorization = signature }
+      }
+    in
+    match%bind Gql_client.send_zkapp t.l1_uri command with
+    | Ok _ ->
+        t.nonce <- Option.map t.nonce ~f:(fun x -> x + 1) ;
+        return ()
+    | Error (`Failed_request err) when attempt >= t.max_attempts ->
+        failwithf "Failed to send zkapp command: Failed_request %s" err ()
+    | Error (`Graphql_error err) when attempt >= t.max_attempts ->
+        failwithf "Failed to send zkapp command: Graphql_error %s" err ()
+    | Error err ->
+        if
+          String.is_substring
+            (match err with `Graphql_error s -> s | _ -> "")
+            ~substring:"Account_nonce_precondition_unsatisfied"
+        then refresh_nonce t ;
+        retry (attempt + 1) ()
+  in
+  retry 0 ()
+
+let send_zkapp_command t command =
+  Throttle.enqueue t.q (fun () -> process_command t command)
+
+let wait_to_finish t = Throttle.capacity_available t.q

src/app/zeko/sequencer/lib/gql.ml

@@ -1700,7 +1700,8 @@ module Queries = struct
       ~args:Arg.[]
       ~resolve:(fun { ctx = sequencer; _ } () ->
         match
-          Zeko_sequencer.(sequencer.snark_q.previous_committed_ledger_hash)
+          Zeko_sequencer.(
+            sequencer.snark_q.state.previous_committed_ledger_hash)
         with
         | None ->
             return (Ok None)

src/app/zeko/sequencer/lib/gql_client.ml

@@ -56,41 +56,36 @@ let fetch_commited_state uri pk =
 
 let send_zkapp ({ value = uri; _ } : Uri.t Cli_lib.Flag.Types.with_name) command
     =
-  Utils.retry
-    ~f:(fun () ->
-      let q =
-        object
-          method query =
-            String.substr_replace_all ~pattern:"\n" ~with_:" "
-              {|
-                mutation ($input: SendZkappInput!) {
-                  sendZkapp(input: $input){
-                    zkapp {
-                      id
-                      failureReason {
-                        index
-                        failures
-                      }
-                    }
+  let q =
+    object
+      method query =
+        String.substr_replace_all ~pattern:"\n" ~with_:" "
+          {|
+            mutation ($input: SendZkappInput!) {
+              sendZkapp(input: $input){
+                zkapp {
+                  id
+                  failureReason {
+                    index
+                    failures
                   }
-                } 
-              |}
+                }
+              }
+            } 
+          |}
 
-          method variables =
-            `Assoc
-              [ ( "input"
-                , `Assoc
-                    [ ( "zkappCommand"
-                      , Yojson.Safe.to_basic @@ Zkapp_command.to_json command )
-                    ] )
-              ]
-        end
-      in
-      let%bind.Deferred.Result result =
-        Graphql_client.Client.query_json q uri
-      in
-      return (Ok Yojson.Safe.(to_string result)) )
-    ()
+      method variables =
+        `Assoc
+          [ ( "input"
+            , `Assoc
+                [ ( "zkappCommand"
+                  , Yojson.Safe.to_basic @@ Zkapp_command.to_json command )
+                ] )
+          ]
+    end
+  in
+  let%bind.Deferred.Result result = Graphql_client.Client.query_json q uri in
+  return (Ok Yojson.Safe.(to_string result))
 
 let fetch_block_height uri =
   let q =