MinaProtocol / mina / 2837

(** 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 Inline_test_quiet_logs

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

  val compile_config : Mina_compile_config.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)

    ~block_window_duration:_ (*TODO remove unused var*) =

  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

    ~transition:cached_initially_validated_transition ~valid_cb =

  let enveloped_initially_validated_transition =

    Cached.peek cached_initially_validated_transition

  in

    Some

  in

  let sender =

    Envelope.Incoming.sender enveloped_initially_validated_transition

  in

    Envelope.Incoming.data enveloped_initially_validated_transition

  in

    let t, _ = initially_validated_transition in

  in

  let state_hash = transition_hash in

  Internal_tracing.with_state_hash state_hash

  @@ fun () ->

    ~metadata:

      [ ( "blockchain_length"

      ] ;

    (let open Deferred.Result.Let_syntax in

    let%bind mostly_validated_transition =

      let open Deferred.Let_syntax in

      match

        Mina_block.Validation.validate_frontier_dependencies

          ~context:(module Context)

          ~root_block:

          ~get_block_by_hash:

            Transition_frontier.(

          initially_validated_transition

      with

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

          let%map () =

              ( Trust_system.Actions.Gossiped_invalid_transition

              , Some

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

                     over the transition frontier root"

                  , metadata ) )

          in

            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

            |> Envelope.Incoming.data

          in

            , _

            , _

            , (`Delta_block_chain, Truth.True delta_state_hashes)

            , _

            , _

            , _ ) ->

              let timeout_duration =

                Option.fold

              in

                ~cached_transition:cached_initially_validated_transition

                ~valid_cb

                ~block_window_duration:compile_config.block_window_duration ;

    in

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

      Protocol_state.previous_state_hash

    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

            (* 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

        ~block_window_duration:compile_config.block_window_duration

    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 ]

       * [ `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 )

         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 )

         Rose_tree.t

         list

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

       , crash buffered

       , unit )

       Writer.t ) ~processed_transition_writer =

  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

                                  ~block_window_duration:

                                    compile_config.block_window_duration

                              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

                        ~block_window_duration:

                          compile_config.block_window_duration

                    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)

                  (`Block transition, `Valid_cb valid_cb) ->

                  process_transition ~transition ~valid_cb ) ) )

let%test_module "Transition_handler.Processor tests" =

  ( module struct

    open Async

    open Pipe_lib

    let () =

      Backtrace.elide := false ;

      Printexc.record_backtrace true ;

    let proof_level = precomputed_values.proof_level

    let constraint_constants = precomputed_values.constraint_constants

    let time_controller = Block_time.Controller.basic ~logger

    let compile_config = Mina_compile_config.For_unit_tests.t

    let verifier =

            ~conf_dir:None

            ~commit_id:"not specified for unit tests" () )

    module Context = struct

      let logger = logger

      let precomputed_values = precomputed_values

      let constraint_constants = constraint_constants

      let consensus_constants = precomputed_values.consensus_constants

      let compile_config = compile_config

    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 ;

                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 ) ) )