1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
type t = {
state : Store.t;
db : Distributed_db.t;
block_validator : Block_validator.t;
chain_validator_limits : Shell_limits.chain_validator_limits;
peer_validator_limits : Shell_limits.peer_validator_limits;
block_validator_limits : Shell_limits.block_validator_limits;
prevalidator_limits : Shell_limits.prevalidator_limits;
start_testchain : bool;
valid_block_input : Store.Block.t Lwt_watcher.input;
chains_input : (Chain_id.t * bool) Lwt_watcher.input;
active_chains : Chain_validator.t Chain_id.Table.t;
}
let create state db peer_validator_limits block_validator_limits
block_validator_kind prevalidator_limits chain_validator_limits
~start_testchain =
let open Lwt_result_syntax in
let*! block_validator =
Block_validator.create
block_validator_limits
db
block_validator_kind
~start_testchain
in
let valid_block_input = Lwt_watcher.create_input () in
let chains_input = Lwt_watcher.create_input () in
return
{
state;
db;
start_testchain;
block_validator;
block_validator_limits;
prevalidator_limits;
peer_validator_limits;
chain_validator_limits;
valid_block_input;
chains_input;
active_chains = Chain_id.Table.create 7;
}
let activate v ~start_prevalidator ~validator_process chain_store =
let open Lwt_syntax in
let chain_id = Store.Chain.chain_id chain_store in
let* () = Validator_event.(emit activate_chain) chain_id in
match Chain_id.Table.find v.active_chains chain_id with
| Some chain -> return_ok chain
| None ->
Chain_validator.create
~start_prevalidator
~start_testchain:v.start_testchain
~active_chains:v.active_chains
~block_validator_process:validator_process
v.peer_validator_limits
v.prevalidator_limits
v.block_validator
v.valid_block_input
v.chains_input
v.db
chain_store
v.chain_validator_limits
let get {active_chains; _} chain_id =
let open Result_syntax in
match Chain_id.Table.find active_chains chain_id with
| Some nv -> return nv
| None -> tzfail (Validation_errors.Inactive_chain chain_id)
let get_active_chains {active_chains; _} =
let l = Chain_id.Table.fold (fun c _ acc -> c :: acc) active_chains [] in
List.rev l
let read_block store h =
let open Lwt_option_syntax in
let*! chain_stores = Store.all_chain_stores store in
List.find_map_s
(fun chain_store ->
let* b = Store.Block.read_block_opt chain_store h in
let id = Store.Chain.chain_id chain_store in
return (id, b))
chain_stores
let db h =
let open Lwt_option_syntax in
let* chain_id, block = read_block (Distributed_db.store db) h in
let = Store.Block.header block in
return (chain_id, header)
let validate_block v ?(force = false) ?chain_id bytes operations =
let open Lwt_result_syntax in
let hash = Block_hash.hash_bytes [bytes] in
match Block_header.of_bytes bytes with
| None -> failwith "Cannot parse block header."
| Some block ->
if not (Clock_drift.is_not_too_far_in_the_future block.shell.timestamp)
then failwith "Block in the future."
else
let* nv =
match chain_id with
| None -> (
let*! o = read_block_header v.db block.shell.predecessor in
match o with
| None ->
failwith
"Unknown predecessor (%a), cannot inject the block."
Block_hash.pp_short
block.shell.predecessor
| Some (chain_id, _bh) -> Lwt.return (get v chain_id))
| Some chain_id -> (
let* nv = Lwt.return (get v chain_id) in
if force then return nv
else
let*! b =
Distributed_db.Block_header.known
(Chain_validator.chain_db nv)
block.shell.predecessor
in
match b with
| true -> return nv
| false ->
failwith
"Unknown predecessor (%a), cannot inject the block."
Block_hash.pp_short
block.shell.predecessor)
in
let validation =
Chain_validator.validate_block nv ~force hash block operations
in
return (hash, validation)
let shutdown {active_chains; block_validator; _} =
let open Lwt_syntax in
let chain_validator_jobs =
List.of_seq
@@ Seq.map
(fun (id, nv) ->
let* () = Validator_event.(emit shutdown_chain_validator) id in
Chain_validator.shutdown nv)
(Chain_id.Table.to_seq active_chains)
in
let* () = Lwt.join chain_validator_jobs in
let* () = Validator_event.(emit shutdown_block_validator) () in
Block_validator.shutdown block_validator
let watcher {valid_block_input; _} = Lwt_watcher.create_stream valid_block_input
let chains_watcher {chains_input; _} = Lwt_watcher.create_stream chains_input
let inject_operation v ?chain_id ~force op =
let open Lwt_result_syntax in
let inject_operation_on nv ~handle_missing_prevalidator =
match Chain_validator.prevalidator nv with
| Some pv -> Prevalidator.inject_operation pv ~force op
| None -> handle_missing_prevalidator
in
let handle_missing_prevalidator =
failwith "Prevalidator is not running, cannot inject the operation."
in
match chain_id with
| None -> (
let*! o = read_block_header v.db op.Operation.shell.branch in
match o with
| None ->
if force then
Chain_id.Table.iter_es
(fun _chain_id chain ->
inject_operation_on
chain
~handle_missing_prevalidator:return_unit)
v.active_chains
else
failwith
"Unknown branch (%a), cannot inject the operation."
Block_hash.pp_short
op.shell.branch
| Some (chain_id, _bh) ->
let*? nv = get v chain_id in
inject_operation_on nv ~handle_missing_prevalidator)
| Some chain_id ->
let*? nv = get v chain_id in
inject_operation_on nv ~handle_missing_prevalidator
let distributed_db {db; _} = db