Revision 50ad02fa2a56f1d3541a6d8b3c8a6a93160d4f54 authored by Albin Coquereau on 05 April 2024, 09:32:54 UTC, committed by Albin Coquereau on 05 April 2024, 09:32:54 UTC
1 parent 9a71363
chain_validator.ml
(*****************************************************************************)
(* *)
(* Open Source License *)
(* Copyright (c) 2018 Dynamic Ledger Solutions, Inc. <contact@tezos.com> *)
(* Copyright (c) 2018-2021 Nomadic Labs. <contact@nomadic-labs.com> *)
(* *)
(* Permission is hereby granted, free of charge, to any person obtaining a *)
(* copy of this software and associated documentation files (the "Software"),*)
(* to deal in the Software without restriction, including without limitation *)
(* the rights to use, copy, modify, merge, publish, distribute, sublicense, *)
(* and/or sell copies of the Software, and to permit persons to whom the *)
(* Software is furnished to do so, subject to the following conditions: *)
(* *)
(* The above copyright notice and this permission notice shall be included *)
(* in all copies or substantial portions of the Software. *)
(* *)
(* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR*)
(* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, *)
(* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL *)
(* THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER*)
(* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING *)
(* FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER *)
(* DEALINGS IN THE SOFTWARE. *)
(* *)
(*****************************************************************************)
open Chain_validator_worker_state
module Name = struct
type t = Chain_id.t
let encoding = Chain_id.encoding
let base = ["validator"; "chain"]
let pp = Chain_id.pp_short
let equal = Chain_id.equal
end
module Request = struct
include Request
type (_, _) t =
| Validated : {
peer : P2p_peer_id.t option;
(* The peer who sent the block if it was not injected locally. *)
block : Store.Block.t;
resulting_context_hash : Context_hash.t;
}
-> (update, error trace) t
| Notify_branch : P2p_peer.Id.t * Block_locator.t -> (unit, Empty.t) t
| Notify_head :
P2p_peer.Id.t * Block_hash.t * Block_header.t * Mempool.t
-> (unit, Empty.t) t
| Disconnection : P2p_peer.Id.t -> (unit, Empty.t) t
let view (type a b) (req : (a, b) t) : view =
match req with
| Validated {block; _} -> Hash (Store.Block.hash block)
| Notify_branch (peer_id, _) -> PeerId peer_id
| Notify_head (peer_id, _, _, _) -> PeerId peer_id
| Disconnection peer_id -> PeerId peer_id
end
module Types = struct
type parameters = {
parent : Name.t option;
(* inherit bootstrap status from parent chain validator *)
db : Distributed_db.t;
chain_store : Store.chain_store;
block_validator : Block_validator.t;
block_validator_process : Block_validator_process.t;
global_valid_block_input : Store.Block.t Lwt_watcher.input;
global_chains_input : (Chain_id.t * bool) Lwt_watcher.input;
start_prevalidator : bool;
prevalidator_limits : Shell_limits.prevalidator_limits;
peer_validator_limits : Shell_limits.peer_validator_limits;
limits : Shell_limits.chain_validator_limits;
metrics : Shell_metrics.Chain_validator.t;
}
type state = {
parameters : parameters;
chain_db : Distributed_db.chain_db;
(* This state should be updated everytime a block is validated or
when we receive a message [Current_head] or [Current_branch]
with a known head. Because the chain validator does not handle
directly these messages, this is done through callbacks. *)
synchronisation_state : Synchronisation_heuristic.Bootstrapping.t;
valid_block_input : Store.Block.t Lwt_watcher.input;
new_head_input : (Block_hash.t * Block_header.t) Lwt_watcher.input;
mutable child : (state * (unit -> unit Lwt.t (* shutdown *))) option;
prevalidator : Prevalidator.t option ref;
active_peers : (Peer_validator.t, Empty.t) P2p_peer.Error_table.t;
}
let is_bootstrapped (state : state) =
Synchronisation_heuristic.Bootstrapping.is_bootstrapped
state.synchronisation_state
end
module Events = Chain_validator_events
module Worker = Worker.MakeSingle (Name) (Request) (Types)
open Types
type t = Worker.infinite Worker.queue Worker.t
let table = Worker.create_table Queue
let shutdown w = Worker.shutdown w
let shutdown_child nv active_chains =
let open Lwt_syntax in
Option.iter_s
(fun ({parameters = {chain_store; global_chains_input; _}; _}, shutdown) ->
let test_chain_id = Store.Chain.chain_id chain_store in
Lwt_watcher.notify global_chains_input (test_chain_id, false) ;
Chain_id.Table.remove active_chains test_chain_id ;
let* r = Store.Chain.shutdown_testchain nv.parameters.chain_store in
match r with
| Error _err ->
(* FIXME *)
Lwt.return_unit
| Ok () ->
let* () = shutdown () in
nv.child <- None ;
Lwt.return_unit)
nv.child
(* Update the synchronisation state and if it is relevant, set the
bootstrapped flag to true. Assume:
- [peer_id] is not us.
- [block] is known as valid. *)
let update_synchronisation_state w block peer_id =
let nv = Worker.state w in
Synchronisation_heuristic.Bootstrapping.update
nv.synchronisation_state
(block.Block_header.shell.timestamp, peer_id)
(* The synchronisation state is updated only for blocks known as
valid. Assume:
- [peer_id] is not us *)
let check_and_update_synchronisation_state w (hash, block) peer_id : unit Lwt.t
=
let open Lwt_syntax in
let nv = Worker.state w in
let* known_valid =
Store.Block.is_known_valid nv.parameters.chain_store hash
in
if known_valid then update_synchronisation_state w block peer_id
else Lwt.return_unit
(* Called for every validated block. *)
let notify_new_block w peer {Block_validator.block; resulting_context_hash} =
let nv = Worker.state w in
Option.iter
(fun id ->
List.assoc ~equal:Chain_id.equal id (Worker.list table)
|> Option.iter (fun w ->
let nv = Worker.state w in
Lwt_watcher.notify nv.valid_block_input block))
nv.parameters.parent ;
Lwt_watcher.notify nv.valid_block_input block ;
Lwt_watcher.notify nv.parameters.global_valid_block_input block ;
Worker.Queue.push_request_now
w
(Validated {peer; block; resulting_context_hash})
let with_activated_peer_validator w peer_id f =
let open Lwt_syntax in
let nv = Worker.state w in
let* pv =
P2p_peer.Error_table.find_or_make nv.active_peers peer_id (fun () ->
let* () = Events.(emit connection) peer_id in
let* pv =
Peer_validator.create
~notify_new_block:(notify_new_block w (Some peer_id))
~notify_termination:(fun _pv ->
P2p_peer.Error_table.remove nv.active_peers peer_id)
nv.parameters.peer_validator_limits
nv.parameters.block_validator
nv.chain_db
peer_id
in
Lwt.return_ok pv)
in
match pv with
| Ok pv -> (
match Peer_validator.status pv with
| Worker_types.Running _ -> f pv
| Worker_types.Closing (_, _)
| Worker_types.Closed (_, _, _)
| Worker_types.Launching _ ->
return_ok_unit)
let may_update_protocol_level chain_store block resulting_context_hash =
let open Lwt_result_syntax in
let* pred = Store.Block.read_predecessor chain_store block in
let prev_proto_level = Store.Block.proto_level pred in
let new_proto_level = Store.Block.proto_level block in
if Compare.Int.(prev_proto_level < new_proto_level) then
let context_index =
Store.context_index (Store.Chain.global_store chain_store)
in
let* resulting_context =
protect (fun () ->
let*! c =
Context_ops.checkout_exn context_index resulting_context_hash
in
return c)
in
let*! new_protocol = Context_ops.get_protocol resulting_context in
let* (module NewProto) = Registered_protocol.get_result new_protocol in
Store.Chain.may_update_protocol_level
chain_store
~pred
~protocol_level:new_proto_level
~expect_predecessor_context:
(NewProto.expected_context_hash = Predecessor_resulting_context)
(block, new_protocol)
else return_unit
let may_switch_test_chain w active_chains spawn_child block =
let open Lwt_result_syntax in
let nv = Worker.state w in
let may_create_child block test_protocol expiration forking_block_hash =
let block_header = Store.Block.header block in
let genesis_hash =
Tezos_context_disk.Context.compute_testchain_genesis forking_block_hash
in
let testchain_id =
Tezos_context_disk.Context.compute_testchain_chain_id genesis_hash
in
let*! activated =
match nv.child with
| None -> Lwt.return_false
| Some (child, _) ->
let child_chain_store = child.parameters.chain_store in
let child_genesis = Store.Chain.genesis child_chain_store in
Lwt.return (Block_hash.equal child_genesis.block genesis_hash)
in
let expired = expiration < block_header.shell.timestamp in
if expired && activated then
let*! () = shutdown_child nv active_chains in
return_unit
else
let chain_store = nv.parameters.chain_store in
let allow_forked_chains =
Store.allow_testchains (Store.Chain.global_store chain_store)
in
if (not allow_forked_chains) || activated || expired then return_unit
else
let* testchain_store =
let*! o =
Store.get_chain_store_opt
(Store.Chain.global_store nv.parameters.chain_store)
testchain_id
in
match o with
| Some test_chain_store -> return test_chain_store
| None ->
let try_init_test_chain cont =
let* forking_block =
Store.Block.read_block chain_store forking_block_hash
in
let bvp = nv.parameters.block_validator_process in
let*! r =
Block_validator_process.init_test_chain
bvp
(Store.Chain.chain_id chain_store)
forking_block
in
match r with
| Ok genesis_header ->
let* testchain =
Store.Chain.fork_testchain
chain_store
~testchain_id
~forked_block:forking_block
~genesis_hash
~genesis_header
~expiration
~test_protocol
in
let testchain_store =
Store.Chain.testchain_store testchain
in
let*! new_genesis_block =
Store.Chain.current_head testchain_store
in
Lwt_watcher.notify
nv.parameters.global_valid_block_input
new_genesis_block ;
Lwt_watcher.notify nv.valid_block_input new_genesis_block ;
return testchain_store
| Error
(Block_validator_errors.Missing_test_protocol
missing_protocol
:: _) ->
let* _ =
Block_validator.fetch_and_compile_protocol
nv.parameters.block_validator
missing_protocol
in
cont ()
| Error _ as error -> Lwt.return error
in
try_init_test_chain @@ fun () ->
try_init_test_chain @@ fun () ->
failwith "Could not retrieve test protocol"
in
(* [spawn_child] is a callback to [create_node]. Thus, it takes care of
global initialization boilerplate (e.g. notifying [global_chains_input],
adding the chain to the correct tables, ...) *)
let* child =
spawn_child
~parent:(Store.Chain.chain_id chain_store)
nv.parameters.start_prevalidator
nv.parameters.peer_validator_limits
nv.parameters.prevalidator_limits
nv.parameters.block_validator
nv.parameters.global_valid_block_input
nv.parameters.global_chains_input
nv.parameters.db
testchain_store
nv.parameters.limits
(* TODO: different limits main/test ? *)
in
nv.child <- Some child ;
return_unit
in
let*! r =
let* v = Store.Block.testchain_status nv.parameters.chain_store block in
match v with
| Not_running, _ ->
let*! () = shutdown_child nv active_chains in
return_unit
| (Forking _ | Running _), None -> return_unit (* only for snapshots *)
| ( (Forking {protocol; expiration; _} | Running {protocol; expiration; _}),
Some forking_block_hash ) ->
may_create_child block protocol expiration forking_block_hash
in
match r with
| Ok () -> Lwt.return_unit
| Error err ->
let*! () = Events.(emit could_not_switch_testchain) err in
Lwt.return_unit
let instantiate_prevalidator parameters set_prevalidator block chain_db =
let open Lwt_syntax in
let* r =
let open Lwt_result_syntax in
let* new_protocol_hash =
Store.Block.protocol_hash parameters.chain_store block
in
let* proto =
Protocol_plugin.proto_with_validation_plugin
~block_hash:(Store.Block.hash block)
new_protocol_hash
in
Prevalidator.create parameters.prevalidator_limits proto chain_db
in
match r with
| Error errs ->
let* () = Events.(emit prevalidator_reinstantiation_failure) errs in
set_prevalidator None ;
Lwt.return_unit
| Ok prevalidator ->
set_prevalidator (Some prevalidator) ;
Lwt.return_unit
let may_flush_or_update_prevalidator parameters event prevalidator chain_db
~prev ~block =
let open Lwt_syntax in
match !prevalidator with
| None -> return_ok_unit
| Some old_prevalidator ->
let prev_proto_level = Store.Block.proto_level prev in
let new_proto_level = Store.Block.proto_level block in
if Compare.Int.(prev_proto_level < new_proto_level) then
(* TODO inject in the new prevalidator the operation
from the previous one. *)
let* () =
instantiate_prevalidator
parameters
(fun new_prevalidator -> prevalidator := new_prevalidator)
block
chain_db
in
let* () = Prevalidator.shutdown old_prevalidator in
return_ok_unit
else
let* live_blocks, live_operations =
Store.Chain.live_blocks parameters.chain_store
in
Prevalidator.flush
old_prevalidator
event
(Store.Block.hash block)
live_blocks
live_operations
let register_garbage_collect_callback w =
let open Lwt_result_syntax in
let nv = Worker.state w in
let chain_store = nv.parameters.chain_store in
let index = Store.(context_index (Chain.global_store chain_store)) in
let gc_lockfile_path =
Naming.(file_path (gc_lockfile (Store.Chain.chain_dir chain_store)))
in
let is_gc_allowed = Context_ops.is_gc_allowed index in
let gc_callback =
if is_gc_allowed then
let gc_synchronous_call block_hash =
let* block = Store.Block.read_block chain_store block_hash in
let* resulting_context_hash =
Store.Block.resulting_context_hash chain_store block
in
let* () =
Block_validator.context_garbage_collection
nv.parameters.block_validator
index
resulting_context_hash
~gc_lockfile_path
in
let kind =
Block_validator_process.kind nv.parameters.block_validator_process
in
(* The callback call will block until the GC is completed: either
using the Irmin function directly in single-process or using
the lockfile through the validator process used to notify the
GC completion. *)
match kind with
| Single_process ->
(* If the GC is running in the same instance, we can directly
wait for its completion. *)
let*! () = Context_ops.wait_gc_completion index in
return_unit
| External_process ->
(* Otherwise, we wait for the lockfile, which is locked by the
external process when the GC starts and released when the
GC ends. *)
let* gc_lockfile =
protect (fun () ->
let*! fd =
Lwt_unix.openfile
gc_lockfile_path
[Unix.O_CREAT; O_RDWR; O_CLOEXEC; O_SYNC]
0o644
in
return fd)
in
Lwt.finalize
(fun () ->
let*! () = Lwt_unix.lockf gc_lockfile Unix.F_RLOCK 0 in
let*! () = Lwt_unix.lockf gc_lockfile Unix.F_ULOCK 0 in
return_unit)
(fun () -> Lwt_unix.close gc_lockfile)
in
Some gc_synchronous_call
else None
in
Store.Chain.register_gc_callback nv.parameters.chain_store gc_callback
let register_split_callback w =
let nv = Worker.state w in
let chain_store = nv.parameters.chain_store in
let index = Store.(context_index (Chain.global_store chain_store)) in
let is_gc_allowed = Context_ops.is_gc_allowed index in
let split =
if is_gc_allowed then
Some
(fun () ->
Block_validator.context_split nv.parameters.block_validator index)
else None
in
Store.Chain.register_split_callback nv.parameters.chain_store split
let may_synchronise_context synchronisation_state chain_store =
if
not
(Synchronisation_heuristic.Bootstrapping.is_bootstrapped
synchronisation_state)
then
(* If the node is bootstrapping, we make sure that the
context is synchronized. *)
let store = Store.Chain.global_store chain_store in
let context_index = Store.context_index store in
Context_ops.sync context_index
else Lwt.return_unit
let on_validation_request w peer start_testchain active_chains spawn_child block
resulting_context_hash =
let open Lwt_result_syntax in
let*! () =
Option.iter_s
(update_synchronisation_state w (Store.Block.header block))
peer
in
let nv = Worker.state w in
let chain_store = nv.parameters.chain_store in
let*! head = Store.Chain.current_head chain_store in
let head_header = Store.Block.header head
and head_hash = Store.Block.hash head
and block_header = Store.Block.header block in
let head_fitness = head_header.shell.fitness in
let new_fitness = block_header.shell.fitness in
let accepted_head = Fitness.(new_fitness > head_fitness) in
if not accepted_head then return Ignored_head
else
let* previous = Store.Chain.set_head chain_store block in
let () =
if is_bootstrapped nv then
Distributed_db.Advertise.current_head nv.chain_db block
in
let* () =
may_update_protocol_level chain_store block resulting_context_hash
in
let*! () =
if start_testchain then
may_switch_test_chain w active_chains spawn_child block
else Lwt.return_unit
in
Lwt_watcher.notify nv.new_head_input (Store.Block.hash block, block_header) ;
let is_head_increment =
Block_hash.equal head_hash block_header.shell.predecessor
in
let event = if is_head_increment then Head_increment else Branch_switch in
let* () =
when_ (not is_head_increment) (fun () ->
Store.Chain.may_update_ancestor_protocol_level chain_store ~head:block)
in
let*! () = may_synchronise_context nv.synchronisation_state chain_store in
let* () =
may_flush_or_update_prevalidator
nv.parameters
event
nv.prevalidator
nv.chain_db
~prev:previous
~block
in
return event
let on_notify_branch w peer_id locator =
let open Lwt_syntax in
let {Block_locator.head_hash; head_header; _} = locator in
let* () =
check_and_update_synchronisation_state w (head_hash, head_header) peer_id
in
with_activated_peer_validator w peer_id (fun pv ->
Peer_validator.notify_branch pv locator ;
return_ok_unit)
let on_notify_head w peer_id (block_hash, header) mempool =
let open Lwt_syntax in
let nv = Worker.state w in
let* () =
check_and_update_synchronisation_state w (block_hash, header) peer_id
in
let* current_head = Store.Chain.current_head nv.parameters.chain_store in
let* (r : (_, Empty.t) result) =
(* To minimize unnecessary calls to notify_head, we do nothing
when the received block is actually the current head or it's
predecessor. *)
if
not
Block_hash.(
Store.Block.hash current_head = block_hash
|| Store.Block.predecessor current_head = block_hash)
then
with_activated_peer_validator w peer_id (fun pv ->
Peer_validator.notify_head pv block_hash header ;
return_ok_unit)
else return_ok_unit
in
match r with
| Ok () -> (
match !(nv.prevalidator) with
| Some prevalidator ->
let* () =
Prevalidator.notify_operations prevalidator peer_id mempool
in
return_ok_unit
| None -> return_ok_unit)
let on_disconnection w peer_id =
let open Lwt_result_syntax in
let nv = Worker.state w in
match P2p_peer.Error_table.find nv.active_peers peer_id with
| None -> return_unit
| Some pv -> (
let*! pv in
match pv with
| Ok pv ->
(* Shutting down the peer validator will indirectly trigger
a disconnection. However, the disconnection can be
effective only after the peer validator is shutdown (see
{!val:Peer_validator.on_close}). This is a defensive
mechanism so that if the disconnection takes too much
time (even though it is not supposed to happen), the
validation is not stuck. In any case, the peer validator
is removed from the active peers (see
{!val:with_activated_peer_validator}). *)
let*! () = Peer_validator.shutdown pv in
return_unit)
let on_request (type a b) w start_testchain active_chains spawn_child
(req : (a, b) Request.t) : (a, b) result Lwt.t =
let nv = Worker.state w in
Prometheus.Counter.inc_one
nv.parameters.metrics.worker_counters.worker_request_count ;
match req with
| Request.Validated {peer; block; resulting_context_hash} ->
on_validation_request
w
peer
start_testchain
active_chains
spawn_child
block
resulting_context_hash
| Request.Notify_branch (peer_id, locator) ->
on_notify_branch w peer_id locator
| Request.Notify_head (peer_id, hash, header, mempool) ->
on_notify_head w peer_id (hash, header) mempool
| Request.Disconnection peer_id -> on_disconnection w peer_id
let collect_proto ~metrics (chain_store, block) =
let open Lwt_syntax in
let* metadata_opt = Store.Block.get_block_metadata_opt chain_store block in
match metadata_opt with
| None -> Lwt.return_unit
| Some metadata ->
let protocol_metadata = Block_repr.block_metadata metadata in
Lwt.catch
(fun () ->
(* Return Noop if the protocol does not exist, and
UndefinedMetric if the plugin for the protocol
does not exist *)
let* (module Metrics_plugin) =
let* protocol = Store.Block.protocol_hash_exn chain_store block in
Protocol_plugin.safe_find_metrics protocol
in
let fitness = Store.Block.fitness block in
let* () =
Metrics_plugin.update_metrics
~protocol_metadata
fitness
(Shell_metrics.Chain_validator.update_proto_metrics_callback
~metrics)
in
Lwt.return_unit)
(fun _ -> Lwt.return_unit)
let on_error (type a b) w st (request : (a, b) Request.t) (errs : b) :
unit tzresult Lwt.t =
let open Lwt_result_syntax in
let nv = Worker.state w in
Prometheus.Counter.inc_one
nv.parameters.metrics.worker_counters.worker_error_count ;
let request_view = Request.view request in
match request with
| Validated _ ->
(* If an error happens here, it means that the request
[Validated] failed. For this request, the payload
associated to the request was validated and therefore is
safe. The handler for such request does some I/Os, and
therefore a failure could be "No space left on device" for
example. If there is an error at this level, it certainly
requires a manual operation from the maintener of the
node. *)
let*! () = Events.(emit request_failure) (request_view, st, errs) in
Lwt.return_error errs
(* We do not crash the worker in the following cases mainly for one
reason: Such request comes from a remote peer. The payload for this
request may contain unsafe data. The current policy with tzresult is
not clear and there might be a non serious error raised as a [tzresult]
to say it was a bad data. If if is the case, we do not want to crash
the worker.
With the current state of the code, it is possible that
this branch is not reachable. This would be possible to
see it if we relax the interface of [tezos-worker] to use
[('a, 'b) result] instead of [tzresult] and if each
request uses its own error type. *)
| Notify_branch _ -> ( match errs with _ -> .)
| Notify_head _ -> ( match errs with _ -> .)
| Disconnection _ -> ( match errs with _ -> .)
let on_completion (type a b) w (req : (a, b) Request.t) (update : a)
request_status =
let open Lwt_syntax in
let nv = Worker.state w in
Prometheus.Counter.inc_one
nv.parameters.metrics.worker_counters.worker_completion_count ;
match req with
| Request.Validated {block; _} ->
let level = Store.Block.level block in
let () =
Shell_metrics.Worker.update_timestamps
nv.parameters.metrics.worker_timestamps
request_status ;
match update with
| Ignored_head ->
Prometheus.Counter.inc_one nv.parameters.metrics.ignored_head_count
| Branch_switch ->
Prometheus.Counter.inc_one nv.parameters.metrics.branch_switch_count ;
Prometheus.Gauge.set
nv.parameters.metrics.head_level
(Int32.to_float level) ;
Shell_metrics.Chain_validator.update_proto (fun () ->
collect_proto
~metrics:nv.parameters.metrics
(nv.parameters.chain_store, block))
| Head_increment ->
Prometheus.Counter.inc_one
nv.parameters.metrics.head_increment_count ;
Prometheus.Gauge.set
nv.parameters.metrics.head_level
(Int32.to_float level) ;
Shell_metrics.Chain_validator.update_proto (fun () ->
collect_proto
~metrics:nv.parameters.metrics
(nv.parameters.chain_store, block))
in
let fitness = Store.Block.fitness block in
let block_hash = Request.Hash (Store.Block.hash block) in
let timestamp = Store.Block.timestamp block in
let* () =
match update with
| Ignored_head -> Events.(emit ignore_head) (block_hash, level)
| Branch_switch -> Events.(emit branch_switch) (block_hash, level)
| Head_increment ->
if
Synchronisation_heuristic.Bootstrapping.is_bootstrapped
nv.synchronisation_state
then Events.(emit head_increment) (block_hash, level)
else if Int32.rem level 50l = 0l then
(* Display a bootstrapping status message every 50 blocks *)
let now = Time.System.now () in
let block_time = Time.System.of_protocol_exn timestamp in
Chain_validator_events.(emit bootstrap_head_increment)
(level, Ptime.diff now block_time)
else Lwt.return_unit
in
Events.(emit block_info) (timestamp, fitness)
| Request.Notify_head (peer_id, block_hash, _, _) ->
Events.(emit notify_head) (peer_id, block_hash)
| Request.Notify_branch (peer_id, locator) ->
Events.(emit notify_branch) (peer_id, locator.head_hash)
| Request.Disconnection peer_id -> Events.(emit disconnection) peer_id
let on_close w =
let open Lwt_syntax in
let nv = Worker.state w in
let* () = Distributed_db.deactivate nv.chain_db in
let pvs =
P2p_peer.Error_table.fold_promises
(fun _ pv acc ->
(let* r = pv in
match r with
| Error _ -> Lwt.return_unit
| Ok pv -> Peer_validator.shutdown pv)
:: acc)
nv.active_peers
[]
in
Lwt.join
(Option.iter_s Prevalidator.shutdown !(nv.prevalidator)
:: Option.iter_s (fun (_, shutdown) -> shutdown ()) nv.child
:: pvs)
let may_load_protocols parameters =
let open Lwt_result_syntax in
let chain_store = parameters.chain_store in
let global_store = Store.Chain.global_store chain_store in
let*! indexed_protocols = Store.Chain.all_protocol_levels chain_store in
let open Store_types in
Protocol_levels.iter_es
(fun _proto_level {Protocol_levels.protocol; _} ->
if Registered_protocol.mem protocol then return_unit
else
match Store.Protocol.mem global_store protocol with
| false -> return_unit
| true ->
(* Only compile protocols that are on-disk *)
let*! () = Events.(emit loading_protocol protocol) in
trace
(Validation_errors.Cannot_load_protocol protocol)
(let* _ =
Block_validator.fetch_and_compile_protocol
parameters.block_validator
protocol
in
return_unit))
indexed_protocols
type launch_error = error trace
let on_launch w _ parameters =
let open Lwt_result_syntax in
let* () = may_load_protocols parameters in
let valid_block_input = Lwt_watcher.create_input () in
let new_head_input = Lwt_watcher.create_input () in
let notify_branch peer_id locator =
Worker.Queue.push_request_now w (Notify_branch (peer_id, locator))
in
let notify_head peer_id hash header ops =
Worker.Queue.push_request_now w (Notify_head (peer_id, hash, header, ops))
in
let disconnection peer_id =
Worker.Queue.push_request_now w (Disconnection peer_id)
in
let chain_db =
Distributed_db.activate
parameters.db
parameters.chain_store
{notify_branch; notify_head; disconnection}
in
let prevalidator = ref None in
let when_status_changes status =
let*! () = Events.(emit synchronisation_status) status in
Shell_metrics.Chain_validator.update_sync_status
~metrics:parameters.metrics
status ;
match status with
| Synchronisation_heuristic.Synchronised _ ->
if parameters.start_prevalidator then
let*! head = Store.Chain.current_head parameters.chain_store in
instantiate_prevalidator
parameters
(fun new_prevalidator -> prevalidator := new_prevalidator)
head
chain_db
else Lwt.return_unit
| _ -> Lwt.return_unit
in
let synchronisation_state =
Synchronisation_heuristic.Bootstrapping.create
~when_bootstrapped_changes:(fun is_bootstrapped ->
Shell_metrics.Chain_validator.update_bootstrapped
~metrics:parameters.metrics
is_bootstrapped ;
if is_bootstrapped then Events.(emit bootstrapped) ()
else Lwt.return_unit)
~when_status_changes
~threshold:parameters.limits.synchronisation.threshold
~latency:parameters.limits.synchronisation.latency
()
in
let nv =
{
parameters;
chain_db;
valid_block_input;
new_head_input;
synchronisation_state;
active_peers = P2p_peer.Error_table.create 50;
(* TODO use [2 * max_connection] *)
child = None;
prevalidator (* the prevalidator may be instantiated next *);
}
in
(* The synchronisation heuristic must be activated after the
[chain_db] since when [limits.synchronisation.threshold <= 0] ,
it will initialise the prevalidator which requires the
[chain_db]. *)
let*! () =
Synchronisation_heuristic.Bootstrapping.activate synchronisation_state
in
return nv
let metrics = Shell_metrics.Chain_validator.init Name.base
let rec create ~start_testchain ~active_chains ?parent ~block_validator_process
start_prevalidator
(peer_validator_limits : Shell_limits.peer_validator_limits)
(prevalidator_limits : Shell_limits.prevalidator_limits) block_validator
global_valid_block_input global_chains_input db chain_store limits =
let open Lwt_result_syntax in
let spawn_child ~parent enable_prevalidator peer_validator_limits
prevalidator_limits block_validator global_valid_block_input
global_chains_input ddb chain_store limits =
let* w =
create
~start_testchain
~active_chains
~parent
~block_validator_process
enable_prevalidator
peer_validator_limits
prevalidator_limits
block_validator
global_valid_block_input
global_chains_input
ddb
chain_store
limits
in
return (Worker.state w, fun () -> Worker.shutdown w)
in
let module Handlers = struct
type self = t
type nonrec launch_error = launch_error
let on_launch = on_launch
let on_request w = on_request w start_testchain active_chains spawn_child
let on_close = on_close
let on_error = on_error
let on_completion = on_completion
let on_no_request _ = Lwt.return_unit
end in
let chain_id = Store.Chain.chain_id chain_store in
let parameters =
{
parent;
peer_validator_limits;
start_prevalidator;
prevalidator_limits;
block_validator;
block_validator_process;
global_valid_block_input;
global_chains_input;
db;
chain_store;
limits;
metrics = metrics chain_id;
}
in
let* w = Worker.launch table chain_id parameters (module Handlers) in
Chain_id.Table.add active_chains (Store.Chain.chain_id chain_store) w ;
register_garbage_collect_callback w ;
register_split_callback w ;
Lwt_watcher.notify global_chains_input (Store.Chain.chain_id chain_store, true) ;
return w
(** Current block computation *)
let create ~start_prevalidator ~start_testchain ~active_chains
~block_validator_process peer_validator_limits prevalidator_limits
block_validator global_valid_block_input global_chains_input global_db state
limits =
(* hide the optional ?parent *)
create
~start_testchain
~active_chains
~block_validator_process
start_prevalidator
peer_validator_limits
prevalidator_limits
block_validator
global_valid_block_input
global_chains_input
global_db
state
limits
let chain_id w =
let {parameters = {chain_store; _}; _} = Worker.state w in
Store.Chain.chain_id chain_store
let chain_store w =
let {parameters = {chain_store; _}; _} = Worker.state w in
chain_store
let prevalidator w =
let {prevalidator; _} = Worker.state w in
!prevalidator
let chain_db w =
let {chain_db; _} = Worker.state w in
chain_db
let child w =
Option.bind
(Worker.state w).child
(fun ({parameters = {chain_store; _}; _}, _) ->
List.assoc
~equal:Chain_id.equal
(Store.Chain.chain_id chain_store)
(Worker.list table))
let assert_fitness_increases ?(force = false) w distant_header =
let open Lwt_syntax in
let pv = Worker.state w in
let chain_store = Distributed_db.chain_store pv.chain_db in
let* current_head = Store.Chain.current_head chain_store in
fail_when
((not force)
&& Fitness.compare
distant_header.Block_header.shell.fitness
(Store.Block.fitness current_head)
<= 0)
(error_of_fmt "Fitness too low")
let assert_checkpoint w ((hash, _) as block_descr) =
let open Lwt_syntax in
let pv = Worker.state w in
let chain_store = Distributed_db.chain_store pv.chain_db in
let* acceptable = Store.Chain.is_acceptable_block chain_store block_descr in
fail_unless acceptable (Validation_errors.Checkpoint_error (hash, None))
let validate_block w ?force hash block operations =
let open Lwt_result_syntax in
let nv = Worker.state w in
let* () = assert_fitness_increases ?force w block in
let* () = assert_checkpoint w (hash, block.Block_header.shell.level) in
let*! v =
Block_validator.precheck_and_apply
~canceler:(Worker.canceler w)
~notify_new_block:(notify_new_block w None)
~precheck_and_notify:true
nv.parameters.block_validator
nv.chain_db
hash
block
operations
in
match v with
| Valid -> return_unit
| Invalid errs | Unapplicable_after_precheck errs -> Lwt.return_error errs
let bootstrapped w =
let state = Worker.state w in
Synchronisation_heuristic.Bootstrapping.bootstrapped
state.synchronisation_state
let is_bootstrapped w = Types.is_bootstrapped (Worker.state w)
let reconfigure_event_logging w config =
Block_validator_process.reconfigure_event_logging
(Worker.state w).parameters.block_validator_process
config
let force_bootstrapped w b =
let state = Worker.state w in
Synchronisation_heuristic.Bootstrapping.force_bootstrapped
state.synchronisation_state
b
let valid_block_watcher w =
let {valid_block_input; _} = Worker.state w in
Lwt_watcher.create_stream valid_block_input
let new_head_watcher w =
let {new_head_input; _} = Worker.state w in
Lwt_watcher.create_stream new_head_input
let status = Worker.status
let information = Worker.information
let running_workers () = Worker.list table
let pending_requests t = Worker.Queue.pending_requests t
let pending_requests_length t = Worker.Queue.pending_requests_length t
let current_request t = Worker.current_request t
let ddb_information t =
let state = Worker.state t in
let ddb = state.chain_db in
Distributed_db.information ddb
let sync_status w =
let nv = Worker.state w in
Synchronisation_heuristic.Bootstrapping.get_status nv.synchronisation_state
![swh spinner](/static/img/swh-spinner.gif)
Computing file changes ...