MinaProtocol / mina / 2837

open Inline_test_quiet_logs

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

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

  let candidate_consensus_state =

    With_hash.map ~f:Mina_block.consensus_state candidate_transition

  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 ;

        Envelope.Incoming.data incoming_transition

      in

        |> Protocol_state.previous_state_hash

      in

        incoming_transition ;

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

        worth_getting_root t

          ~metadata:

            [ ( "state_hash"

            ; ( "external_transition"

            ] ;

             transition )

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

    (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:Mina_block.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 ~best_seen_transition

    ~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 ( `Block

                               (block :

                                 Mina_block.Validation.initial_valid_with_block

                                 Envelope.Incoming.t )

                           , `Valid_cb valid_cb ) ->

                            Pipe.Drop_on_overflow.bootstrap_sync_ledger) ;

                        Mina_block.handle_dropped_transition ?valid_cb

                          ~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

                  ~preferred:

                               None

                               Some r ) )

                  ~root_sync_ledger ~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 ->

                        |> Mina_block.Validation.block_with_hash

                      in

                      @@ Consensus.Hooks.select

                           ~context:(module Context)

                           ~existing:root_consensus_state

                           ~candidate:

                                transition ) )

                in

                  ~metadata:[] ;

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

                        ~compare:

                             ~f:

                                  Envelope.Incoming.data )

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

            ~conf_dir:None

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

    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

                      , `Vallid_cb None ) )

              in

            List.map branch

              ~f:

                      Transition_frontier.Breadcrumb.validated_transition ) )

          in

            |> List.map

                 ~f:

                      Envelope.Incoming.data )

          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

           ~best_seen_transition:None

           ~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 :

          Mina_block.initial_valid_block Envelope.Incoming.t list ) =

            ~f:(fun max_acc incoming_transition ->

                Envelope.Incoming.data incoming_transition

              in

              let%map () =

                  Mina_numbers.Length.(

                  ~error:

                       "The blocks are not sorted in increasing order" )

              in

          : Mina_block.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

              in

              let%map actual_staged_ledger =

                Staged_ledger.of_scan_state_pending_coinbases_and_snarked_ledger

                  ~scan_state ~logger ~verifier ~constraint_constants

                  ~snarked_ledger ~snarked_local_state ~expected_merkle_root

                  ~pending_coinbases ~get_state

              in

    (*

    let%test_unit "if we see a new transition that is better than the \

                   transition that we are syncing from, than we should \

                   retarget our root" =

      Quickcheck.test ~trials:1

        Fake_network.Generator.(

          gen ~max_frontier_length

            [ fresh_peer

            ; peer_with_branch ~frontier_branch_size:max_frontier_length

            ; peer_with_branch

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

        ~f:(fun fake_network ->

          let [me; weaker_chain; stronger_chain] =

            fake_network.peer_networks

          in

          let transition_reader, transition_writer =

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

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

          in

          Envelope.Incoming.wrap