MinaProtocol / mina / 2956

open Core

open Async

open Mina_base

module Ledger = Mina_ledger.Ledger

module Sync_ledger = Mina_ledger.Sync_ledger

open Mina_state

open Pipe_lib.Strict_pipe

open Network_peer

module Transition_cache = Transition_cache

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

type Structured_log_events.t += Bootstrap_complete

type t =

  { context : (module CONTEXT)

  ; trust_system : Trust_system.t

  ; verifier : Verifier.t

  ; mutable best_seen_transition : Mina_block.initial_valid_block

  ; mutable current_root : Mina_block.initial_valid_block

  ; network : Mina_networking.t

  ; mutable num_of_root_snarked_ledger_retargeted : int

  }

type time = Time.Span.t

let time_to_yojson span =

type opt_time = time option

let opt_time_to_yojson = function

      time_to_yojson time

      `Null

(** An auxiliary data structure for collecting various metrics for boostrap controller. *)

  }

[@@deriving to_yojson]

let time_deferred deferred =

  let%map result = deferred in

let worth_getting_root ({ context = (module Context); _ } as t) candidate =

    include Context

    let logger =

        [ ( "selection_context"

          , `String "Bootstrap_controller.worth_getting_root" )

        ]

  end in

  Consensus.Hooks.equal_select_status `Take

  @@ Consensus.Hooks.select

       ~context:(module Context)

       ~existing:

       ~candidate

let received_bad_proof ({ context = (module Context); _ } as t) host e =

  Trust_system.(

    record t.trust_system logger host

      Actions.

        ( Violated_protocol

        , Some

            ( "Bad ancestor proof: $error"

let done_syncing_root root_sync_ledger =

let should_sync ~root_sync_ledger t candidate_state =

(** Update [Synced_ledger]'s target and [best_seen_transition] and [current_root] accordingly. *)

let start_sync_job_with_peer ~sender ~root_sync_ledger

    ({ context = (module Context); _ } as t) peer_best_tip peer_root =

  let%bind () =

    Trust_system.(

        Actions.

          ( Fulfilled_request

          , Some ("Received verified peer root and best tip", []) ))

  in

  t.current_root <- peer_root ;

  let blockchain_state =

  in

    blockchain_state |> Blockchain_state.staged_ledger_hash

  in

    blockchain_state |> Blockchain_state.snarked_ledger_hash

  in

  @@

  match

    Sync_ledger.Db.new_goal root_sync_ledger

      ~data:

        , sender

        , expected_staged_ledger_hash )

  with

      t.num_of_root_snarked_ledger_retargeted <-

        t.num_of_root_snarked_ledger_retargeted + 1 ;

      `Syncing_new_snarked_ledger

      `Updating_root_transition

      `Ignored

let to_consensus_state h =

(** For each transition, this function would compare it with the existing one.

    If the incoming transition is better, then download the merkle list from

    that transition to its root and verify it. If we get a better root than

    the existing one, then reset the Sync_ledger's target by calling

    [start_sync_job_with_peer] function. *)

let on_transition ({ context = (module Context); _ } as t) ~sender

    ~root_sync_ledger ~genesis_constants candidate_header =

  let candidate_consensus_state =

    With_hash.map ~f:to_consensus_state candidate_header

  in

  else

    match%bind

           ~f:State_hash.State_hashes.state_hash )

    with

          !"Could not get the proof of the root transition from the network: \

            $error" ;

        match%bind

            ~context:(module Context)

            ~verifier:t.verifier ~genesis_constants candidate_consensus_state

            peer_root_with_proof.data

        with

            else

(** A helper function that wraps the calls to Sync_ledger and iterate through

    incoming transitions, add those to the transition_cache and calls

    [on_transition] function. *)

let sync_ledger ({ context = (module Context); _ } as t) ~preferred

    ~root_sync_ledger ~transition_graph ~sync_ledger_reader ~genesis_constants =

  let query_reader = Sync_ledger.Db.query_reader root_sync_ledger in

    response_writer ;

        match b_or_h with

      in

      let previous_state_hash =

        |> Protocol_state.previous_state_hash

      in

        (transition_cache_element, vc) ;

      (* TODO: Efficiently limiting the number of green threads in #1337 *)

        worth_getting_root t

          ~metadata:

            [ ( "state_hash"

            ; ( "header"

            ] ;

             header_with_hash )

let external_transition_compare ~context:(module Context : CONTEXT) =

    Fn.compose Protocol_state.consensus_state Mina_block.Header.protocol_state

  in

    (fun existing candidate ->

      (* To prevent the logger to spam a lot of messsages, the logger input is set to null *)

        State_hash.equal

        Consensus.Hooks.equal_select_status `Keep

        @@ Consensus.Hooks.select ~context:(module Context) ~existing ~candidate

    ~f:(With_hash.map ~f:get_consensus_state)

(** The entry point function for bootstrap controller. When bootstrap finished

    it would return a transition frontier with the root breadcrumb and a list

    of transitions collected during bootstrap.

    Bootstrap controller would do the following steps to contrust the

    transition frontier:

    1. Download the root snarked_ledger.

    2. Download the scan state and pending coinbases.

    3. Construct the staged ledger from the snarked ledger, scan state and

       pending coinbases.

    4. Synchronize the consensus local state if necessary.

    5. Close the old frontier and reload a new one from disk.

 *)

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

    ~consensus_local_state ~transition_reader ~preferred_peers ~persistent_root

    ~persistent_frontier ~initial_root_transition ~catchup_mode =

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

        Precomputed_values.genesis_constants precomputed_values

      in

      let rec loop previous_cycles =

        let sync_ledger_reader, sync_ledger_writer =

          create ~name:sync_ledger_pipe

            (Buffered

               ( `Capacity 50

               , `Overflow

                   (Drop_head

                      (fun (b_or_h, `Valid_cb valid_cb) ->

                          match b_or_h with

                        in

                        Mina_metrics.(

                            Pipe.Drop_on_overflow.bootstrap_sync_ledger) ;

                        Mina_block.handle_dropped_transition ?valid_cb hash

                          ~pipe_name:sync_ledger_pipe ~logger ) ) ) )

        in

             ~f:Fn.id ) ;

          |> Mina_block.Validation.reset_staged_ledger_diff_validation

        in

          { network

          ; context = (module Context)

          ; trust_system

          ; verifier

          ; best_seen_transition = initial_root_transition

          ; current_root = initial_root_transition

          ; num_of_root_snarked_ledger_retargeted = 0

          }

        in

        let transition_graph = Transition_cache.create () in

          Transition_frontier.Persistent_root.create_instance_exn

            persistent_root

        in

          Transition_frontier.Persistent_root.Instance.snarked_ledger

            temp_persistent_root_instance

        in

        (* step 1. download snarked_ledger *)

        let%bind sync_ledger_time, (hash, sender, expected_staged_ledger_hash) =

            (let root_sync_ledger =

               Sync_ledger.Db.create temp_snarked_ledger ~logger ~trust_system

             in

                  ~transition_graph ~sync_ledger_reader ~genesis_constants ) ;

             (* We ignore the resulting ledger returned here since it will always

                * be the same as the ledger we started with because we are syncing

                * a db ledger. *)

        in

        Mina_metrics.(

        (* step 2. Download scan state and pending coinbases. *)

        let%bind ( staged_ledger_data_download_time

                 , staged_ledger_construction_time

                 , staged_ledger_aux_result ) =

          let%bind ( staged_ledger_data_download_time

                   , staged_ledger_data_download_result ) =

              (Mina_networking

                 sender.peer_id hash )

          in

              Deferred.return (staged_ledger_data_download_time, None, Error err)

              ( scan_state

              , expected_merkle_root

              , pending_coinbases

              , protocol_states ) -> (

              let%map staged_ledger_construction_result =

                    let received_staged_ledger_hash =

                      Staged_ledger_hash.of_aux_ledger_and_coinbase_hash

                        expected_merkle_root pending_coinbases

                    in

                      ~metadata:

                        [ ( "expected_staged_ledger_hash"

                              expected_staged_ledger_hash )

                        ; ( "received_staged_ledger_hash"

                              received_staged_ledger_hash )

                        ]

                      "Comparing $expected_staged_ledger_hash to \

                       $received_staged_ledger_hash" ;

                    let%bind new_root =

                      t.current_root

                      |> Mina_block.Validation

                           `This_block_belongs_to_a_detached_subtree

                           (`Staged_ledger_already_materialized

                             received_staged_ledger_hash )

                               "received faulty scan state from peer" )

                    in

                      List.map protocol_states

                        ~f:(With_hash.of_data ~hash_data:Protocol_state.hashes)

                    in

                    let%bind protocol_states =

                        scan_state ~protocol_states

                    in

                      protocol_states

                      |> State_hash.Map.of_alist_exn

                    in

                          let new_state_hash =

                            State_hash.With_state_hashes.state_hash

                          in

                            ~metadata:

                              ]

                            "Protocol state (for scan state transactions) for \

                             $state_hash not found when bootstrapping to the \

                             new root $new_root" ;

                              %{sexp:State_hash.t} not found when \

                              bootstrapping to the new root \

                              %{sexp:State_hash.t}"

                            hash new_state_hash

                    in

                    (* step 3. Construct staged ledger from snarked ledger, scan state

                       and pending coinbases. *)

                    (* Construct the staged ledger before constructing the transition

                     * frontier in order to verify the scan state we received.

                     * TODO: reorganize the code to avoid doing this twice (#3480) *)

                    let open Deferred.Let_syntax in

                    let%map staged_ledger_construction_time, construction_result

                        =

                        (let open Deferred.Let_syntax in

                        let temp_mask =

                          Ledger.of_database temp_snarked_ledger

                        in

                        let%map result =

                          Staged_ledger

                          .of_scan_state_pending_coinbases_and_snarked_ledger

                            ~logger

                            ~snarked_local_state:

                              Mina_block.(

                            ~verifier ~constraint_constants ~scan_state

                            ~snarked_ledger:temp_mask ~expected_merkle_root

                            ~pending_coinbases ~get_state

                        in

                              temp_mask

                            : Ledger.unattached_mask ) ;

                        Result.map result

                          ~f:

                               ( scan_state

                               , pending_coinbases

                               , new_root

                               , protocol_states ) ))

                    in

              in

                  (staged_ledger_data_download_time, None, Error err)

                  ( staged_ledger_data_download_time

                  , Some staged_ledger_construction_time

                  , result ) )

        in

          temp_persistent_root_instance ;

            let%bind () =

              Trust_system.(

                  Actions.

                    ( Outgoing_connection_error

                    , Some

                        ( "Can't find scan state from the peer or received \

                           faulty scan state from the peer."

                        , [] ) ))

            in

              ~metadata:

                ; ( "expected_staged_ledger_hash"

                ]

              "Failed to find scan state for the transition with hash \

               $state_hash from the peer or received faulty scan state: \

               $error. Retry bootstrap" ;

              { cycle_result = "failed to download and construct scan state"

              ; sync_ledger_time

              ; staged_ledger_data_download_time

              ; staged_ledger_construction_time

              ; local_state_sync_required = false

              ; local_state_sync_time = None

              }

            in

            loop (this_cycle :: previous_cycles)

            let%bind () =

              Trust_system.(

                  Actions.

                    ( Fulfilled_request

                    , Some ("Received valid scan state from peer", []) ))

            in

            let consensus_state =

              |> Protocol_state.consensus_state

            in

            (* step 4. Synchronize consensus local state if necessary *)

            let%bind ( local_state_sync_time

                     , (local_state_sync_required, local_state_sync_result) ) =

                ( match

                    Consensus.Hooks.required_local_state_sync

                      ~constants:precomputed_values.consensus_constants

                      ~consensus_state ~local_state:consensus_local_state

                  with

                      ~metadata:

                        [ ( "local_state"

                              consensus_local_state )

                        ; ( "consensus_state"

                              consensus_state )

                        ]

                      "Not synchronizing consensus local state" ;

                    let%map result =

                        ~context:(module Context)

                        ~local_state:consensus_local_state ~trust_system

                        ~glue_sync_ledger:

                        sync_jobs

                    in

            in

                  "Local state sync failed: $error. Retry bootstrap" ;

                  { cycle_result = "failed to synchronize local state"

                  ; sync_ledger_time

                  ; staged_ledger_data_download_time

                  ; staged_ledger_construction_time

                  ; local_state_sync_required

                  ; local_state_sync_time = Some local_state_sync_time

                  }

                in

                loop (this_cycle :: previous_cycles)

                (* step 5. Close the old frontier and reload a new one from disk. *)

                let new_root_data : Transition_frontier.Root_data.Limited.t =

                  Transition_frontier.Root_data.Limited.create

                    ~scan_state ~pending_coinbase ~protocol_states

                in

                let%bind () =

                    persistent_frontier ~root_data:new_root_data

                    ~genesis_state_hash:

                         precomputed_values.protocol_state_with_hashes )

                in

                (* TODO: lazy load db in persistent root to avoid unecessary opens like this *)

                let%map new_frontier =

                  let fail msg =

                      ( "failed to initialize transition frontier after \

                         bootstrapping: " ^ msg )

                  in

                    ~context:(module Context)

                    ~retry_with_fresh_db:false ~verifier ~consensus_local_state

                    ~persistent_root ~persistent_frontier ~catchup_mode ()

                      frontier

                      fail msg

                      fail

                        "bootstrap still required (indicates logical error in \

                         code)"

                      fail "persistent frontier was malformed"

                      fail

                        "this should not happen, because we just reset the \

                         snarked_ledger"

                in

                  Transition_cache.data transition_graph

                in

                  Logger.extend logger

                    [ ( "context"

                      , `String "Filter collected transitions in bootstrap" )

                    ]

                in

                  Transition_frontier.(

                in

                let filtered_collected_transitions =

                  List.filter collected_transitions

                    ~f:(fun (incoming_transition, _) ->

                        |> Transition_cache.header_with_hash

                      in

                      @@ Consensus.Hooks.select

                           ~context:(module Context)

                           ~existing:root_consensus_state

                           ~candidate:

                                ~f:

                                     Mina_block.Header.protocol_state )

                                transition ) )

                in

                  ~metadata:[] ;

                  O1trace.sync_thread "sorting_collected_transitions" (fun () ->

                        ~compare:

                             ~f:(fun (x, _) ->

                             (external_transition_compare

                                ~context:(module Context) ) ) )

                in

                  { cycle_result = "success"

                  ; sync_ledger_time

                  ; staged_ledger_data_download_time

                  ; staged_ledger_construction_time

                  ; local_state_sync_required

                  ; local_state_sync_time = Some local_state_sync_time

                  }

                in

                ( this_cycle :: previous_cycles

                , (new_frontier, sorted_filtered_collected_transitions) ) )

      in

        ~metadata:

          ; ( "bootstrap_stats"

          ] ;

      result )

let%test_module "Bootstrap_controller tests" =

  ( module struct

    open Pipe_lib

    let max_frontier_length =

    let () =

      (* Disable log messages from best_tip_diff logger. *)

        ~transport:

             ( module struct

               type t = unit

             end )

             () )

        ()

    let trust_system =

      let s = Trust_system.null () in

    let proof_level = precomputed_values.proof_level

    let constraint_constants = precomputed_values.constraint_constants

    let compile_config = Mina_compile_config.For_unit_tests.t

    module Context = struct

      let logger = logger

      let precomputed_values = precomputed_values

      let constraint_constants =

        Genesis_constants.For_unit_tests.Constraint_constants.t

      let consensus_constants = precomputed_values.consensus_constants

      let compile_config = compile_config

    end

    let verifier =

            () )

    module Genesis_ledger = (val precomputed_values.genesis_ledger)

    let downcast_transition ~sender transition =

        |> Mina_block.Validation.reset_staged_ledger_diff_validation

      in

        ~sender:(Envelope.Sender.Remote sender)

    let downcast_breadcrumb ~sender breadcrumb =

    let make_non_running_bootstrap ~genesis_root ~network =

        genesis_root

        |> Mina_block.Validation.reset_staged_ledger_diff_validation

      in

      ; trust_system

      ; verifier

      ; best_seen_transition = transition

      ; current_root = transition

      ; network

      ; num_of_root_snarked_ledger_retargeted = 0

      }

    let%test_unit "Bootstrap controller caches all transitions it is passed \

                   through the transition_reader" =

      Quickcheck.test ~trials:1

        (let open Quickcheck.Generator.Let_syntax in

        (* we only need one node for this test, but we need more than one peer so that mina_networking does not throw an error *)

        let%bind fake_network =

          Fake_network.Generator.(

              ~compile_config [ fresh_peer; fresh_peer ]

              ~use_super_catchup:false)

        in

        let%map make_branch =

            ~verifier

            branch_size

        in

        let branch =

          Async.Thread_safe.block_on_async_exn (fun () ->

        in

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

          let genesis_root =

            Transition_frontier.(

            |> Mina_block.Validated.remember

          in

            Pipe_lib.Strict_pipe.create ~name:"sync_ledger_reader" Synchronous

          in

            make_non_running_bootstrap ~genesis_root ~network:me.network

          in

          let root_sync_ledger =

            Sync_ledger.Db.create

              ~logger ~trust_system

          in

                sync_ledger bootstrap ~root_sync_ledger ~transition_graph

                  ~preferred:[] ~sync_ledger_reader

                  ~genesis_constants:Genesis_constants.For_unit_tests.t

              in

              let%bind () =

                      ( `Block

                      , `Valid_cb None ) )

              in

            List.map branch

              ~f:

                   (With_hash.map ~f:Mina_block.header)

                         Transition_frontier.Breadcrumb.validated_transition ) ) )

          in

            |> List.map ~f:(fun (x, _) ->

          in

            module T = struct

              [@@deriving sexp]

              let compare = external_transition_compare ~context:(module Context)

            end

            include Comparable.Make (T)

          end in

    let run_bootstrap ~timeout_duration ~my_net ~transition_reader =

      let time_controller = Block_time.Controller.basic ~logger in

      let persistent_root =

        Transition_frontier.persistent_root my_net.state.frontier

      in

        Transition_frontier.persistent_frontier my_net.state.frontier

      in

        Transition_frontier.(

      in

      let%bind () =

      in

        (run

           ~context:(module Context)

           ~trust_system ~verifier ~network:my_net.network ~preferred_peers:[]

           ~consensus_local_state:my_net.state.consensus_local_state

           ~transition_reader ~persistent_root ~persistent_frontier

           ~catchup_mode:`Normal ~initial_root_transition )

    let assert_transitions_increasingly_sorted ~root

        (incoming_transitions : Transition_cache.element list) =

      in

            ~f:(fun max_acc incoming_transition ->

                Transition_cache.header_with_hash

              in

                |> Consensus.Data.Consensus_state.blockchain_length

              in

              let open Result.Let_syntax in

              let%map () =

                  ~error:

                       "The blocks are not sorted in increasing order" )

              in

          : Mina_block.Header.t )

    let%test_unit "sync with one node after receiving a transition" =

        Fake_network.Generator.(

            ~use_super_catchup:false ~compile_config

            [ fresh_peer

            ; peer_with_branch

                ~frontier_branch_size:((max_frontier_length * 2) + 2)

            ])

        ~f:(fun fake_network ->

          let transition_reader, transition_writer =

            Pipe_lib.Strict_pipe.create ~name:(__MODULE__ ^ __LOC__)

              (Buffered (`Capacity 10, `Overflow (Drop_head ignore)))

          in

            Envelope.Incoming.wrap

              ~data:

              ~sender:(Envelope.Sender.Remote peer_net.peer)

          in

          Pipe_lib.Strict_pipe.Writer.write transition_writer

            (`Block block, `Valid_cb None) ;

            Async.Thread_safe.block_on_async_exn (fun () ->

                  ~my_net ~transition_reader )

          in

            sorted_external_transitions ;

            ~expect:

              ) )

    let%test_unit "reconstruct staged_ledgers using \

                   of_scan_state_and_snarked_ledger" =

           ~max_length:max_frontier_length ~size:max_frontier_length () )

        ~f:(fun frontier ->

          @@ fun () ->

            ~f:(fun breadcrumb ->

                Transition_frontier.Breadcrumb.staged_ledger breadcrumb

              in

              in

                |> Ledger.of_database

              in

                |> Blockchain_state.snarked_local_state

              in

                    Ok protocol_state

                    Or_error.errorf

                        %{sexp:State_hash.t} not found"

                      hash

              in

              let pending_coinbases =

                Staged_ledger.pending_coinbase_collection staged_ledger