MinaProtocol / mina / 1239

(** This module contains the transition processor. The transition processor is

 *  the thread in which transitions are attached the to the transition frontier.

 *

 *  Two types of data are handled by the transition processor: validated external transitions

 *  with precomputed state hashes (via the block producer and validator pipes)

 *  and breadcrumb rose trees (via the catchup pipe).

 *)

open Core_kernel

open Async_kernel

open Pipe_lib.Strict_pipe

open Mina_base

open Mina_state

open Cache_lib

open Mina_block

open Network_peer

module type CONTEXT = sig

  val logger : Logger.t

  val precomputed_values : Precomputed_values.t

  val constraint_constants : Genesis_constants.Constraint_constants.t

  val consensus_constants : Consensus.Constants.t

end

(* TODO: calculate a sensible value from postake consensus arguments *)

let catchup_timeout_duration (precomputed_values : Precomputed_values.t) =

    ( (precomputed_values.genesis_constants.protocol.delta + 1)

      * precomputed_values.constraint_constants.block_window_duration_ms

let cached_transform_deferred_result ~transform_cached ~transform_result cached

    =

(* add a breadcrumb and perform post processing *)

let add_and_finalize ~logger ~frontier ~catchup_scheduler

    ~processed_transition_writer ~only_if_present ~time_controller ~source

    ~valid_cb cached_breadcrumb ~(precomputed_values : Precomputed_values.t) =

  let breadcrumb =

  in

  let consensus_constants = precomputed_values.consensus_constants in

  let transition =

    Transition_frontier.Breadcrumb.validated_transition breadcrumb

  in

    ~metadata:

      [ ( "state_hash"

      ]

  @@ fun () ->

  let%map () =

      let parent_hash = Transition_frontier.Breadcrumb.parent_hash breadcrumb in

              transition frontier: %{sexp: State_hash.t}"

            parent_hash ;

  in

      ()

      let transition_time =

        |> Consensus.Data.Consensus_state.consensus_time

      in

        Block_time.diff

             transition_time )

      in

    Writer.write processed_transition_writer

      (`Transition transition, `Source source, `Valid_cb valid_cb) ;

let process_transition ~context:(module Context : CONTEXT) ~trust_system

    ~verifier ~get_completed_work ~frontier ~catchup_scheduler

    ~processed_transition_writer ~time_controller ~block_or_header ~valid_cb

    ?transaction_pool_proxy =

  in

  let header, transition_hash, transition_receipt_time, sender, validation =

    match block_or_header with

        let env = Cached.peek cached_env in

        , v )

        let h, v = Envelope.Incoming.data env in

        , v )

  in

  let parent_hash =

  in

    Transition_frontier.(

  in

  let state_hash = transition_hash in

  Internal_tracing.with_state_hash state_hash

  @@ fun () ->

    ~metadata:

      [ ( "blockchain_length"

      ] ;

      ~metadata:[ ("reason", `String "Not_selected_over_frontier_root") ] ;

      ( Trust_system.Actions.Gossiped_invalid_transition

      , Some

          ( "The transition with hash $state_hash was not selected over the \

             transition frontier root"

          , metadata ) )

  in

  match block_or_header with

      let header_with_hash, _ = Envelope.Incoming.data env in

        Mina_block.Validation.validate_frontier_dependencies

          ~context:(module Context)

          ~root_consensus_state ~is_block_in_frontier ~to_header:ident

      with

            ~header_with_hash ;

          handle_not_selected ()

            ~metadata:[ ("reason", `String "Already_in_frontier") ] ;

            "Refusing to process the transition with hash $state_hash because \

             it is already in the transition frontier" ;

      Deferred.ignore_m

      @@

      let open Deferred.Result.Let_syntax in

      let%bind mostly_validated_transition =

        let open Deferred.Let_syntax in

        let initially_validated_transition =

          |> Envelope.Incoming.data

        in

          Mina_block.Validation.validate_frontier_dependencies

            ~context:(module Context)

            ~root_consensus_state ~is_block_in_frontier

            ~to_header:Mina_block.header initially_validated_transition

        with

            let (_ : Mina_block.initial_valid_block Envelope.Incoming.t) =

              Cached.invalidate_with_failure

                cached_initially_validated_transition

            in

              ~metadata:[ ("reason", `String "Already_in_frontier") ] ;

              "Refusing to process the transition with hash $state_hash \

               because is is already in the transition frontier" ;

              Cached.invalidate_with_failure

                cached_initially_validated_transition

            in

              , _

              , _

              , (`Delta_block_chain, Mina_stdlib.Truth.True delta_state_hashes)

              , _

              , _

              , _ ) ->

                let timeout_duration =

                  Option.fold

                in

                  ~cached_transition:cached_initially_validated_transition

                  ~valid_cb ;

      in

      (* TODO: only access parent in transition frontier once (already done in call to validate dependencies) #2485 *)

      let%bind parent_breadcrumb =

        match Transition_frontier.find frontier parent_hash with

            (* Unexpected case: if it happens, there is a bug somewhere. Still,

               we should attempt to handle it gracefully. *)

              ~metadata:[ ("reason", `String "Parent_missing_from_frontier") ] ;

              "Refusing to process the transition with hash $state_hash \

               because parent is missing from the transition frontier \

               (unexpected case, there is likely a bug somewhere)" ;

              Cached.invalidate_with_failure

                cached_initially_validated_transition

            in

      in

      let%bind breadcrumb =

          ~transform_cached:(fun _ ->

              ~verifier ~get_completed_work ~trust_system

              ~transition_receipt_time ~sender:(Some sender)

              ~parent:parent_breadcrumb ~transition:mostly_validated_transition

              ?transaction_pool_proxy (* TODO: Can we skip here? *) () )

          ~transform_result:(function

                Internal_tracing.with_state_hash state_hash

                @@ fun () ->

                  ~metadata:

                  "Error while building breadcrumb in the transition handler \

                   processor: $error" ;

                Internal_tracing.with_state_hash state_hash

                @@ fun () ->

                  ~metadata:[ ("reason", `String "Fatal error") ] ;

                Deferred.return (Ok breadcrumb) )

      in

          Transition_frontier_controller.breadcrumbs_built_by_processor) ;

      let%map.Deferred result =

          ~processed_transition_writer ~only_if_present:false ~time_controller

          ~source:`Gossip breadcrumb ~precomputed_values ~valid_cb

      in

      Result.return result

let run ~context:(module Context : CONTEXT) ~verifier ~trust_system

    ~time_controller ~frontier ~get_completed_work

    ~(primary_transition_reader :

       ( [ `Block of

           ( Mina_block.initial_valid_block Envelope.Incoming.t

           , State_hash.t )

           Cached.t

         | `Header of Mina_block.initial_valid_header Envelope.Incoming.t ]

       * [ `Valid_cb of Mina_net2.Validation_callback.t option ] )

       Reader.t )

    ~(producer_transition_reader : Transition_frontier.Breadcrumb.t Reader.t)

    ~(clean_up_catchup_scheduler : unit Ivar.t) ~catchup_job_writer

    ~(catchup_breadcrumbs_reader :

       ( ( (Transition_frontier.Breadcrumb.t, State_hash.t) Cached.t

         * Mina_net2.Validation_callback.t option )

         Mina_stdlib.Rose_tree.t

         list

       * [ `Ledger_catchup of unit Ivar.t | `Catchup_scheduler ] )

       Reader.t )

    ~(catchup_breadcrumbs_writer :

       ( ( (Transition_frontier.Breadcrumb.t, State_hash.t) Cached.t

         * Mina_net2.Validation_callback.t option )

         Mina_stdlib.Rose_tree.t

         list

         * [ `Ledger_catchup of unit Ivar.t | `Catchup_scheduler ]

       , crash buffered

       , unit )

       Writer.t ) ~processed_transition_writer ?transaction_pool_proxy =

  let catchup_scheduler =

    Catchup_scheduler.create ~logger ~precomputed_values ~verifier ~trust_system

      ~frontier ~time_controller ~catchup_job_writer ~catchup_breadcrumbs_writer

      ~clean_up_signal:clean_up_catchup_scheduler

  in

  let add_and_finalize =

    add_and_finalize ~frontier ~catchup_scheduler ~processed_transition_writer

      ~time_controller ~precomputed_values

  in

  let process_transition =

    process_transition

      ~context:(module Context)

      ~get_completed_work ~trust_system ~verifier ~frontier ~catchup_scheduler

      ~processed_transition_writer ~time_controller

  in

  O1trace.background_thread "process_blocks" (fun () ->

        (* It is fine to skip the cache layer on blocks produced by this node

           * because it is extraordinarily unlikely we would write an internal bug

           * triggering this case, and the external case (where we received an

           * identical external transition from the network) can happen iff there

           * is another node with the exact same private key and view of the

           * transaction pool. *)

                  Transition_frontier_controller.transitions_being_processed) ;

            ~f:(fun (cb, catchup_breadcrumbs_callback) ->

        ]

        ~f:(fun msg ->

          O1trace.thread "transition_handler_processor" (fun () ->

                  (breadcrumb_subtrees, subsequent_callback_action) -> (

                  ( match%map

                        ~f:(fun subtree ->

                            subtree

                            (* It could be the case that by the time we try and

                               * add the breadcrumb, it's no longer relevant when

                               * we're catching up *) ~f:(fun (b, valid_cb) ->

                                Frontier_base.Breadcrumb.state_hash

                              in

                              let%map result =

                                  ~source:`Catchup ~valid_cb b

                              in

                              @@ fun () ->

                                    ~metadata:

                                      [ ( "reason"

                                      ]

                              result ) )

                    with

                      ()

                      List.iter breadcrumb_subtrees ~f:(fun tree ->

                            ~f:(fun (cached_breadcrumb, _vc) ->

                                Cached.invalidate_with_failure cached_breadcrumb

                              in

                        "Error, failed to attach all catchup breadcrumbs to \

                         transition frontier: $error"

                  )

                  >>| fun () ->

                      if Ivar.is_full decrement_signal then

                      () )

                  let state_hash =

                    |> Mina_block.Validated.state_hash

                  in

                  @@ fun () ->

                  in

                    ~name:"accepted_transition_local_latency"

                       transition_time ) ;

                  let%map () =

                    match%map

                        ~source:`Internal breadcrumb ~valid_cb:None

                    with

                          ~metadata:

                          ~metadata:

                          "Error, failed to attach produced breadcrumb to \

                           transition frontier: $error" ;

                          Cached.invalidate_with_failure breadcrumb

                        in

                  in

                    Gauge.dec_one

                      Transition_frontier_controller.transitions_being_processed)

                ->

                  process_transition ~block_or_header ~valid_cb

                    ?transaction_pool_proxy ) ) )

let%test_module "Transition_handler.Processor tests" =

  ( module struct

    open Async

    open Pipe_lib

    let () =

      Backtrace.elide := false ;

      Printexc.record_backtrace true ;

    let () =

      (* Disable log messages from best_tip_diff logger. *)

        ~transport:

             ( module struct

               type t = unit

             end )

             () )

        ()

    let proof_level = precomputed_values.proof_level

    let constraint_constants = precomputed_values.constraint_constants

    let time_controller = Block_time.Controller.basic ~logger

    let verifier =

            () )

    module Context = struct

      let logger = logger

      let precomputed_values = precomputed_values

      let constraint_constants = constraint_constants

      let consensus_constants = precomputed_values.consensus_constants

    end

    let downcast_breadcrumb breadcrumb =

        |> Mina_block.Validation.reset_staged_ledger_diff_validation

      in

    let%test_unit "adding transitions whose parents are in the frontier" =

      let branch_size = 10 in

      let max_length = frontier_size + branch_size in

      Quickcheck.test ~trials:4

           ~verifier ~max_length ~frontier_size ~branch_size () )

        ~f:(fun (frontier, branch) ->

                  Strict_pipe.create

                    (Buffered

                       (`Capacity branch_size, `Overflow (Drop_head ignore)) )

                in

                  Strict_pipe.create

                    (Buffered

                       (`Capacity branch_size, `Overflow (Drop_head ignore)) )

                in

                  Strict_pipe.create (Buffered (`Capacity 1, `Overflow Crash))

                in

                  Strict_pipe.create (Buffered (`Capacity 1, `Overflow Crash))

                in

                  Strict_pipe.create

                    (Buffered

                       (`Capacity branch_size, `Overflow (Drop_head ignore)) )

                in

                  Unprocessed_transition_cache.create ~logger

                    ~cache_exceptions:true

                in

                run

                  ~context:(module Context)

                  ~trust_system ~clean_up_catchup_scheduler ~frontier

                  ~primary_transition_reader:valid_transition_reader

                  ~producer_transition_reader ~catchup_job_writer

                  ~catchup_breadcrumbs_reader ~catchup_breadcrumbs_writer

                  ~processed_transition_writer ?transaction_pool_proxy:None ;

                List.iter branch ~f:(fun breadcrumb ->

                      |> Unprocessed_transition_cache.register_exn cache

                    in

                      (`Block b, `Valid_cb None) ) ;

                match%map

                    time_controller

                       ~init:branch

                       ~f:(fun

                            remaining_breadcrumbs

                            (`Transition newly_added_transition, _, _)

                          ->

                           ( match remaining_breadcrumbs with

                                    next_expected_breadcrumb )

                                    newly_added_transition ) ;

                                 ~metadata:

                                   [ ( "height"

                                     , `Int

                                         ( newly_added_transition

                                         |> Consensus.Data.Consensus_state

                                   ]

                                 "transition of $height passed processor" ;

                               `Stop false ) ) )

                with

                    failwith "test timed out"

                    failwith "pipe closed unexpectedly"

                    x ) ) )