MinaProtocol / mina / 1239

open Core

open Async

open Mina_base

open Mina_state

open Pipe_lib.Strict_pipe

open Network_peer

open Mina_stdlib

module Ledger = Mina_ledger.Ledger

module Root_ledger = Mina_ledger.Root

module Sync_ledger = Mina_ledger.Sync_ledger

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 ledger_sync_config : Syncable_ledger.daemon_config

  val proof_cache_db : Proof_cache_tag.cache_db

  val signature_kind : Mina_signature_kind.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

  }

(** An auxiliary data structure for collecting various metrics for bootstrap 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 Consensus_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.Root.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 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" ;

        let pcd =

          peer_root_with_proof.data

               ~f:

                 (Mina_block.write_all_proofs_to_disk ~signature_kind

                    ~proof_cache_db )

          |> Proof_carrying_data.map_proof

               ~f:

                 (Tuple2.map_snd

                    ~f:

                      (Mina_block.write_all_proofs_to_disk ~signature_kind

                         ~proof_cache_db ) )

        in

        match%bind

            ~verify:

                 ~context:(module Context)

                 ~verifier:t.verifier candidate_consensus_state )

            pcd

        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 =

  let query_reader = Sync_ledger.Root.query_reader root_sync_ledger in

    response_writer ;

    ~f:(fun (b_or_h, `Valid_cb vc) ->

        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"

            ] ;

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

let download_snarked_ledger ~trust_system ~preferred_peers ~transition_graph

    ~sync_ledger_reader ~context t temp_snarked_ledger =

    (let root_sync_ledger =

       Sync_ledger.Root.create temp_snarked_ledger ~context ~trust_system

     in

          ~transition_graph ~sync_ledger_reader ) ;

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

let handle_scan_state_and_aux ~logger ~expected_staged_ledger_hash

    ~temp_snarked_ledger ~verifier ~constraint_constants ~signature_kind

    ~proof_cache_db t

    ( scan_state_uncached

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

            ; ( "received_staged_ledger_hash"

            ]

          "Comparing $expected_staged_ledger_hash to \

           $received_staged_ledger_hash" ;

        let%bind new_root =

          t.current_root

               `This_block_belongs_to_a_detached_subtree

               (`Staged_ledger_already_materialized received_staged_ledger_hash)

        in

          List.map protocol_states

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

        in

          Staged_ledger.Scan_state.write_all_proofs_to_disk ~signature_kind

            ~proof_cache_db scan_state_uncached

        in

        let%bind protocol_states =

            ~protocol_states

        in

          protocol_states

          |> State_hash.Map.of_alist_exn

        in

              let new_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 = Root_ledger.as_masked temp_snarked_ledger in

            (* [of_scan_state_pending_coinbases_and_snarked_ledger] is called here

             * to verify the scan state we received. We ignore the resulting staged

             * ledger (from [temp_mask]). Later we call [Transition_frontier.load]

             * which will indirectly invoke

             * [of_scan_state_pending_coinbases_and_snarked_ledger_unchecked]

             * repeating the same computation (except for proof verification

             * which is skipped). Chain of calls:

             * [Transition_frontier.load] -> .. -> [load_from_persistence_and_start]

             * -> [load_full_frontier] -> [construct_staged_ledger_at_root] ->

             * -> [of_scan_state_pending_coinbases_and_snarked_ledger_unchecked]

             *)

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

            in

                : Ledger.unattached_mask ) ;

            Result.map result

              ~f:

                   (scan_state, pending_coinbases, new_root, protocol_states) ))

        in

  in

      (None, Error err)

      (Some staged_ledger_construction_time, result)

(** Run one bootstrap cycle *)

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

    ~network ~consensus_local_state ~network_transition_pipe ~preferred_peers

    ~persistent_root ~persistent_frontier ~initial_root_transition ~catchup_mode

    previous_cycles =

  (* The short-lived pipe allocated here will be closed

     when a follow-up pipe is allocated: in the next cycle of bootstrap

     or when controll is passed to the transition frontier controller.staged_ledger_construction_time

     Because [Choosable_synchronous_pipe.t] is used, no data will be lost: any

     successful read is followed by mom-blocking handling, and if the read/write

     pair is not consumed, it will continue in a follow-up pipe allocated in the

     next call to [Swappable.swap_reader].

  *)

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

      ~context:(module Context)

      ~trust_system ~preferred_peers ~transition_graph ~sync_ledger_reader t

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

           t.network sender.peer_id hash )

    in

        Deferred.return (staged_ledger_data_download_time, (None, Error err))

        let%map res =

            ~temp_snarked_ledger ~verifier ~constraint_constants ~signature_kind

            ~proof_cache_db t result

        in

  in

    temp_persistent_root_instance ;

      let%map () =

        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

      `Repeat (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_state"

                        consensus_state )

                  ]

                "Not synchronizing consensus local state" ;

              let%map result =

                  ~context:(module Context)

                  ~local_state:consensus_local_state ~trust_system

                  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

          Deferred.return (`Repeat (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 unnecessary 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

            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

          `Finished

            ( this_cycle :: previous_cycles

            , (new_frontier, sorted_filtered_collected_transitions) ) )

(** 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 ~network_transition_pipe ~preferred_peers

    ~persistent_root ~persistent_frontier ~initial_root_transition ~catchup_mode

    =

  let run_cycle =

    run_cycle

      ~context:(module Context : CONTEXT)

      ~trust_system ~verifier ~network ~consensus_local_state

      ~network_transition_pipe ~preferred_peers ~persistent_root

      ~persistent_frontier ~initial_root_transition ~catchup_mode

  in

  O1trace.thread "bootstrap"

  @@ fun () ->

  let%map time_elapsed, (cycles, result) =

  in

    ~metadata:

      ; ( "bootstrap_stats"

      ] ;

  result

let%test_module "Bootstrap_controller tests" =

  ( module struct

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

        ~compile_config:Mina_compile_config.For_unit_tests.t

        ~max_subtree_depth:None ~default_subtree_depth:None ()

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

          ~compile_config:Mina_compile_config.For_unit_tests.t

          ~max_subtree_depth:None ~default_subtree_depth:None ()

      let signature_kind = Mina_signature_kind.Testnet

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

              ~ledger_sync_config [ fresh_peer; fresh_peer ])

        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.Choosable_synchronous_pipe.create ()

          in

            make_non_running_bootstrap ~genesis_root ~network:me.network

          in

          let root_sync_ledger =

            Sync_ledger.Root.create

              ~context:(module Context)

              ~trust_system

          in

                sync_ledger bootstrap ~root_sync_ledger ~transition_graph

                  ~preferred:[] ~sync_ledger_reader

              in

              let%bind sync_ledger_writer' =

                  ~f:(fun sync_ledger_writer breadcrumb ->

                      ( `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 ~network_transition_pipe =

      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

           ~network_transition_pipe ~persistent_root ~persistent_frontier

           ~catchup_mode:`Super ~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.(

            ~ledger_sync_config

            [ fresh_peer

            ; peer_with_branch

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

            ])

        ~f:(fun fake_network ->

          let block =

            Envelope.Incoming.wrap

              ~data:

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

          in

          let network_transition_pipe =

            Swappable.create ~name:(__MODULE__ ^ __LOC__)

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

          in

            Async.Thread_safe.block_on_async_exn (fun () ->

                  ~my_net ~network_transition_pipe )

          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

                |> Root_ledger.as_masked

              in

                |> Blockchain_state.snarked_local_state

              in

                    Ok protocol_state

                    Or_error.errorf

                        %{sexp:State_hash.t} not found"

                      hash