From 7bd2dfeb4e1e400f3ff02fcfd1e35e4251450f75 Mon Sep 17 00:00:00 2001 From: Alain Mebsout Date: Thu, 16 Mar 2023 14:39:25 +0100 Subject: [PATCH 1/2] Injector: save last seen head on disk This is needed to compute reorganizations autonomously. --- src/lib_injector/disk_persistence.mli | 17 ++++++++++- src/lib_injector/injector_functor.ml | 44 +++++++++++++++++++++------ 2 files changed, 50 insertions(+), 11 deletions(-) diff --git a/src/lib_injector/disk_persistence.mli b/src/lib_injector/disk_persistence.mli index 0245cbe641b8..67364a200c7f 100644 --- a/src/lib_injector/disk_persistence.mli +++ b/src/lib_injector/disk_persistence.mli @@ -1,7 +1,8 @@ (*****************************************************************************) (* *) (* Open Source License *) -(* Copyright (c) 2022 Nomadic Labs, *) +(* Copyright (c) 2023 Nomadic Labs, *) +(* Copyright (c) 2023 Functori, *) (* *) (* Permission is hereby granted, free of charge, to any person obtaining a *) (* copy of this software and associated documentation files (the "Software"),*) @@ -31,6 +32,20 @@ type error += | Unix_error of Unix.error | Decoding_error of Data_encoding.Binary.read_error +(** [maybe_read_value ~warn filename encoding] reads the value with [encoding] + from the file [filename]. It returns [None] if it fails to read a value and + will emit a warning with the [warn] function in this case. *) +val maybe_read_value : + warn:(string -> error trace -> unit Lwt.t) -> + string -> + 'a Data_encoding.t -> + 'a option Lwt.t + +(** [write_value filenam encoding v] write the value [v] to the file [filenmae] + in binary form following the [encoding]. *) +val write_value : + string -> 'a Data_encoding.t -> 'a -> (unit, error trace) result Lwt.t + (** Signature for hash tables with additional information *) module type H = sig include Hashtbl.SeededS diff --git a/src/lib_injector/injector_functor.ml b/src/lib_injector/injector_functor.ml index 3b734933b92c..4a1e1426a126 100644 --- a/src/lib_injector/injector_functor.ml +++ b/src/lib_injector/injector_functor.ml @@ -222,6 +222,9 @@ struct retention_period : int; (** Number of blocks for which the injector keeps the included information. *) + mutable last_seen_head : (Block_hash.t * int32) option; + (** Last L1 head that the injector has seen (used to compute + reorganizations). *) } module Event = struct @@ -236,6 +239,24 @@ struct let emit3 e state x y z = emit e (state.signer.pkh, state.tags, x, y, z) end + let last_head_encoding = Data_encoding.(tup2 Block_hash.encoding int32) + + let read_last_head ~data_dir ~warn = + Disk_persistence.maybe_read_value + ~warn + (Filename.concat data_dir "last_seen_head") + last_head_encoding + + let set_last_head state head = + let open Lwt_result_syntax in + let+ () = + Disk_persistence.write_value + (Filename.concat state.save_dir "last_seen_head") + last_head_encoding + head + in + state.last_seen_head <- Some head + let init_injector cctxt ~data_dir state ~retention_period ~signer strategy tags = let open Lwt_result_syntax in @@ -245,13 +266,11 @@ struct let filter op_proj op = Tags.mem (Parameters.operation_tag (op_proj op)) tags in - let warn_unreadable = - (* Warn of corrupted files but don't fail *) - Some - (fun file error -> - Event.(emit corrupted_operation_on_disk) - (signer.pkh, tags, file, error)) + (* Warn of corrupted files but don't fail *) + let warn file error = + Event.(emit corrupted_operation_on_disk) (signer.pkh, tags, file, error) in + let warn_unreadable = Some warn in let emit_event_loaded kind nb = Event.(emit loaded_from_disk) (signer.pkh, tags, nb, kind) in @@ -313,7 +332,7 @@ struct emit_event_loaded "included_in_blocks" @@ Included_in_blocks.length included_in_blocks in - + let*! last_seen_head = read_last_head ~data_dir ~warn in return { cctxt = injector_context (cctxt :> #Client_context.full); @@ -326,6 +345,7 @@ struct included = {included_operations; included_in_blocks}; state; retention_period; + last_seen_head; } (** Add an operation to the pending queue corresponding to the signer for this @@ -875,7 +895,7 @@ struct that are in the alternative branch of the reorganization and then registers the effect of the new branch (the newly included operation and confirmed operations). *) - let on_new_tezos_head state (head_hash, head_level) + let on_new_tezos_head state ((head_hash, head_level) as head) (reorg : (Block_hash.t * int32) Reorg.t) = let open Lwt_result_syntax in let*! () = Event.(emit1 new_tezos_head) state head_hash in @@ -893,8 +913,12 @@ struct (* Head is already included in the reorganization, so no need to process it separately. *) let confirmed_level = Int32.sub head_level (Int32.of_int confirmations) in - if confirmed_level >= 0l then register_confirmed_level state confirmed_level - else return_unit + let* () = + if confirmed_level >= 0l then + register_confirmed_level state confirmed_level + else return_unit + in + set_last_head state head (* The request {Request.Inject} triggers an injection of the operations the pending queue. *) -- GitLab From cd51aad557193ec79a2a1d5c67e8d508defb880f Mon Sep 17 00:00:00 2001 From: Alain Mebsout Date: Wed, 15 Feb 2023 10:02:38 +0100 Subject: [PATCH 2/2] Injector: monitors new Tezos heads on its own The rollup node does not need to notify the injector of new heads anymore. --- src/lib_injector/disk_persistence.mli | 2 +- src/lib_injector/injector_events.ml | 8 + src/lib_injector/injector_functor.ml | 214 +++++++++++------- src/lib_injector/injector_functor.mli | 3 +- src/lib_injector/injector_sigs.ml | 18 +- src/lib_injector/injector_worker_types.ml | 27 +-- src/lib_injector/injector_worker_types.mli | 7 +- .../lib_sc_rollup_node/daemon.ml | 17 -- src/proto_alpha/lib_sc_rollup_node/daemon.ml | 17 -- 9 files changed, 162 insertions(+), 151 deletions(-) diff --git a/src/lib_injector/disk_persistence.mli b/src/lib_injector/disk_persistence.mli index 67364a200c7f..5afb8935465d 100644 --- a/src/lib_injector/disk_persistence.mli +++ b/src/lib_injector/disk_persistence.mli @@ -41,7 +41,7 @@ val maybe_read_value : 'a Data_encoding.t -> 'a option Lwt.t -(** [write_value filenam encoding v] write the value [v] to the file [filenmae] +(** [write_value filename encoding v] write the value [v] to the file [filename] in binary form following the [encoding]. *) val write_value : string -> 'a Data_encoding.t -> 'a -> (unit, error trace) result Lwt.t diff --git a/src/lib_injector/injector_events.ml b/src/lib_injector/injector_events.ml index d5cdbecd1b3c..acc4a3d134b1 100644 --- a/src/lib_injector/injector_events.ml +++ b/src/lib_injector/injector_events.ml @@ -37,6 +37,14 @@ module Make let section = Parameters.events_section + let monitoring_error = + declare_1 + ~section + ~name:"monitoring_error" + ~msg:"error (ignored) in monitoring: {error}" + ~level:Warning + ("error", trace_encoding) + let declare_1 ~name ~msg ~level ?pp1 enc1 = declare_3 ~section diff --git a/src/lib_injector/injector_functor.ml b/src/lib_injector/injector_functor.ml index 4a1e1426a126..68a596876926 100644 --- a/src/lib_injector/injector_functor.ml +++ b/src/lib_injector/injector_functor.ml @@ -1,7 +1,8 @@ (*****************************************************************************) (* *) (* Open Source License *) -(* Copyright (c) 2022 Nomadic Labs, *) +(* Copyright (c) 2023 Nomadic Labs, *) +(* Copyright (c) 2023 Functori, *) (* *) (* Permission is hereby granted, free of charge, to any person obtaining a *) (* copy of this software and associated documentation files (the "Software"),*) @@ -203,6 +204,7 @@ struct type state = { cctxt : Client_context.full; (** The client context which is used to perform the injections. *) + l1_ctxt : Layer_1.t; (** Monitoring of L1 heads. *) signer : signer; (** The signer for this worker. *) tags : Tags.t; (** The tags of this worker, for both informative and identification @@ -257,8 +259,8 @@ struct in state.last_seen_head <- Some head - let init_injector cctxt ~data_dir state ~retention_period ~signer strategy - tags = + let init_injector cctxt l1_ctxt ~data_dir state ~retention_period ~signer + strategy tags = let open Lwt_result_syntax in let* signer = get_signer cctxt signer in let data_dir = Filename.concat data_dir "injector" in @@ -336,6 +338,7 @@ struct return { cctxt = injector_context (cctxt :> #Client_context.full); + l1_ctxt; signer; tags; strategy; @@ -890,15 +893,22 @@ struct else return_unit) state.included.included_in_blocks - (** [on_new_tezos_head state head reorg] is called when there is a new Tezos - head (with a potential reorganization [reorg]). It first reverts any blocks - that are in the alternative branch of the reorganization and then registers - the effect of the new branch (the newly included operation and confirmed - operations). *) - let on_new_tezos_head state ((head_hash, head_level) as head) - (reorg : (Block_hash.t * int32) Reorg.t) = + (** [on_new_tezos_head state head] is called when there is a new Tezos + head. It first reverts any blocks that are in the alternative branch of + the reorganization and then registers the effect of the new branch (the + newly included operation and confirmed operations). *) + let on_new_tezos_head state ((head_hash, head_level) as head) = let open Lwt_result_syntax in let*! () = Event.(emit1 new_tezos_head) state head_hash in + let* reorg = + match state.last_seen_head with + | None -> return {Reorg.no_reorg with new_chain = [head]} + | Some last_head -> + Layer_1.get_tezos_reorg_for_new_head + state.l1_ctxt + (`Head last_head) + head + in let* () = List.iter_es (fun (removed_block, _) -> @@ -929,6 +939,7 @@ struct type parameters = { cctxt : Client_context.full; + l1_ctxt : Layer_1.t; data_dir : string; state : Parameters.state; retention_period : int; @@ -962,17 +973,19 @@ struct (* The execution of the request handler is protected to avoid stopping the worker in case of an exception. *) protect @@ fun () -> add_pending_operation state op - | Request.New_tezos_head (head, reorg) -> - protect @@ fun () -> on_new_tezos_head state head reorg + | Request.New_tezos_head head -> + protect @@ fun () -> on_new_tezos_head state head | Request.Inject -> protect @@ fun () -> on_inject state type launch_error = error trace let on_launch _w signer - Types.{cctxt; data_dir; state; retention_period; strategy; tags} = + Types. + {cctxt; l1_ctxt; data_dir; state; retention_period; strategy; tags} = trace (Step_failed "initialization") @@ init_injector cctxt + l1_ctxt ~data_dir state ~retention_period @@ -1011,10 +1024,88 @@ struct Lwt.return_unit end + let has_tag_in ~tags state = + match tags with + | None -> + (* Not filtering on tags *) + true + | Some tags -> not (Tags.disjoint state.tags tags) + + let delay_strategy state header f = + let open Lwt_syntax in + match state.strategy with + | `Each_block -> f () + | `Delay_block delay_factor -> + let time_until_next_block = + Proto_client.time_until_next_block state.state header + |> Ptime.Span.to_float_s + in + let delay = time_until_next_block *. delay_factor in + if delay <= 0. then f () + else + let promise = + let* () = Event.(emit1 inject_wait) state delay in + let* () = Lwt_unix.sleep delay in + f () + in + ignore promise ; + return_unit + + let inject ?tags ?header () = + let workers = Worker.list table in + let tags = Option.map Tags.of_list tags in + List.iter_p + (fun (_signer, w) -> + let open Lwt_syntax in + let worker_state = Worker.state w in + if has_tag_in ~tags worker_state then + delay_strategy worker_state header @@ fun () -> + let* _pushed = Worker.Queue.push_request w Request.Inject in + return_unit + else Lwt.return_unit) + workers + + let notify_new_tezos_head h = + let open Lwt_syntax in + let workers = Worker.list table in + List.iter_p + (fun (_signer, w) -> + let* (_pushed : bool) = + Worker.Queue.push_request w (Request.New_tezos_head h) + in + return_unit) + workers + + let rec monitor_l1_chain l1_ctxt = + let open Lwt_syntax in + let* res = + Layer_1.iter_heads l1_ctxt @@ fun (head_hash, header) -> + let head = (head_hash, header.shell.level) in + (* Notify all workers of a new Tezos head *) + let* () = notify_new_tezos_head head in + return_ok () + in + (* Ignore errors *) + let* () = + match res with + | Error error -> Internal_event.Simple.emit Event.monitoring_error error + | Ok () -> return_unit + in + monitor_l1_chain l1_ctxt + + let rec async_monitor_l1_chain l1_ctxt = + Lwt.dont_wait + (fun () -> monitor_l1_chain l1_ctxt) + (fun exn -> + Format.eprintf + "Warning: Error in async monitoring (%s), restarting" + (Printexc.to_string exn) ; + async_monitor_l1_chain l1_ctxt) + (* TODO: https://gitlab.com/tezos/tezos/-/issues/2754 Injector worker in a separate process *) let init (cctxt : #Client_context.full) ~data_dir ?(retention_period = 0) - state ~signers = + ?(reconnection_delay = 2.0) state ~signers = let open Lwt_result_syntax in assert (retention_period >= 0) ; let signers_map = @@ -1040,24 +1131,30 @@ struct Signature.Public_key_hash.Map.empty signers in - Signature.Public_key_hash.Map.iter_es - (fun signer (strategy, tags) -> - let+ worker = - Worker.launch - table - signer - { - cctxt = (cctxt :> Client_context.full); - data_dir; - state; - retention_period; - strategy; - tags; - } - (module Handlers) - in - ignore worker) - signers_map + let* l1_ctxt = Layer_1.start ~name:"injector" ~reconnection_delay cctxt in + let* () = + Signature.Public_key_hash.Map.iter_es + (fun signer (strategy, tags) -> + let+ worker = + Worker.launch + table + signer + { + cctxt = (cctxt :> Client_context.full); + l1_ctxt; + data_dir; + state; + retention_period; + strategy; + tags; + } + (module Handlers) + in + ignore worker) + signers_map + in + async_monitor_l1_chain l1_ctxt ; + return_unit let worker_of_signer signer_pkh = match Worker.find_opt table signer_pkh with @@ -1090,60 +1187,9 @@ struct in return operation.hash - let new_tezos_head h reorg = - let open Lwt_syntax in - let workers = Worker.list table in - List.iter_p - (fun (_signer, w) -> - let* (_pushed : bool) = - Worker.Queue.push_request w (Request.New_tezos_head (h, reorg)) - in - return_unit) - workers - - let has_tag_in ~tags state = - match tags with - | None -> - (* Not filtering on tags *) - true - | Some tags -> not (Tags.disjoint state.tags tags) - - let delay_stategy state header f = - let open Lwt_syntax in - match state.strategy with - | `Each_block -> f () - | `Delay_block delay_factor -> - let time_until_next_block = - Proto_client.time_until_next_block state.state header - |> Ptime.Span.to_float_s - in - let delay = time_until_next_block *. delay_factor in - if delay <= 0. then f () - else - let promise = - let* () = Event.(emit1 inject_wait) state delay in - let* () = Lwt_unix.sleep delay in - f () - in - ignore promise ; - return_unit - - let inject ?tags ?header () = - let workers = Worker.list table in - let tags = Option.map Tags.of_list tags in - List.iter_p - (fun (_signer, w) -> - let open Lwt_syntax in - let worker_state = Worker.state w in - if has_tag_in ~tags worker_state then - delay_stategy worker_state header @@ fun () -> - let* _pushed = Worker.Queue.push_request w Request.Inject in - return_unit - else Lwt.return_unit) - workers - let shutdown () = let workers = Worker.list table in + (* Don't shutdown L1 monitoring otherwise worker shutdown hangs *) List.iter_p (fun (_signer, w) -> Worker.shutdown w) workers let op_status_in_worker state l1_hash = diff --git a/src/lib_injector/injector_functor.mli b/src/lib_injector/injector_functor.mli index ab5786a67469..419236989975 100644 --- a/src/lib_injector/injector_functor.mli +++ b/src/lib_injector/injector_functor.mli @@ -1,7 +1,8 @@ (*****************************************************************************) (* *) (* Open Source License *) -(* Copyright (c) 2022 Nomadic Labs, *) +(* Copyright (c) 2023 Nomadic Labs, *) +(* Copyright (c) 2023 Functori, *) (* *) (* Permission is hereby granted, free of charge, to any person obtaining a *) (* copy of this software and associated documentation files (the "Software"),*) diff --git a/src/lib_injector/injector_sigs.ml b/src/lib_injector/injector_sigs.ml index 0cb295f810c0..6774c112e02d 100644 --- a/src/lib_injector/injector_sigs.ml +++ b/src/lib_injector/injector_sigs.ml @@ -1,7 +1,8 @@ (*****************************************************************************) (* *) (* Open Source License *) -(* Copyright (c) 2022 Nomadic Labs, *) +(* Copyright (c) 2023 Nomadic Labs, *) +(* Copyright (c) 2023 Functori, *) (* *) (* Permission is hereby granted, free of charge, to any person obtaining a *) (* copy of this software and associated documentation files (the "Software"),*) @@ -283,11 +284,17 @@ module type S = sig the included information for, must be positive or zero. By default (when [0]), the injector will not keep information longer than necessary. It can be useful to set this value to something [> 0] if we want to retrieve - information about operations included on L1 for a given period. *) + information about operations included on L1 for a given period. + + The injector monitors L1 heads to update the statuses of its operations + accordingly. The argument [reconnection_delay] gives an initial value for + the delay before attempting a reconnection (see {!Layer_1.init}). + *) val init : #Client_context.full -> data_dir:string -> ?retention_period:int -> + ?reconnection_delay:float -> state -> signers:(Signature.public_key_hash * injection_strategy * tag list) list -> unit tzresult Lwt.t @@ -301,13 +308,6 @@ module type S = sig operation -> Inj_operation.Hash.t tzresult Lwt.t - (** Notify the injector of a new Tezos head. The injector marks the operations - appropriately (for instance reverted operations that are part of a - reorganization are put back in the pending queue). When an operation is - considered as {e confirmed}, it disappears from the injector. *) - val new_tezos_head : - Block_hash.t * int32 -> (Block_hash.t * int32) Reorg.t -> unit Lwt.t - (** Trigger an injection of the pending operations for all workers. If [tags] is given, only the workers which have a tag in [tags] inject their pending operations. [header] must be provided for the [`Delay_block] strategy to diff --git a/src/lib_injector/injector_worker_types.ml b/src/lib_injector/injector_worker_types.ml index 3d11e2e53dbc..bd4c61d94b92 100644 --- a/src/lib_injector/injector_worker_types.ml +++ b/src/lib_injector/injector_worker_types.ml @@ -1,7 +1,8 @@ (*****************************************************************************) (* *) (* Open Source License *) -(* Copyright (c) 2022 Nomadic Labs, *) +(* Copyright (c) 2023 Nomadic Labs, *) +(* Copyright (c) 2023 Functori, *) (* *) (* Permission is hereby granted, free of charge, to any person obtaining a *) (* copy of this software and associated documentation files (the "Software"),*) @@ -28,9 +29,7 @@ open Injector_sigs module Request (L1_operation : INJECTOR_OPERATION) = struct type ('a, 'b) t = | Add_pending : L1_operation.t -> (unit, error trace) t - | New_tezos_head : - (Block_hash.t * int32) * (Block_hash.t * int32) Reorg.t - -> (unit, error trace) t + | New_tezos_head : (Block_hash.t * int32) -> (unit, error trace) t | Inject : (unit, error trace) t type view = View : _ t -> view @@ -55,13 +54,11 @@ module Request (L1_operation : INJECTOR_OPERATION) = struct (let block_level = obj2 (req "block" Block_hash.encoding) (req "level" int32) in - obj3 + obj2 (req "request" (constant "new_tezos_head")) - (req "head" block_level) - (req "reorg" (Reorg.encoding block_level))) - (function - | View (New_tezos_head (b, r)) -> Some ((), b, r) | _ -> None) - (fun ((), b, r) -> View (New_tezos_head (b, r))); + (req "head" block_level)) + (function View (New_tezos_head b) -> Some ((), b) | _ -> None) + (fun ((), b) -> View (New_tezos_head b)); case (Tag 2) ~title:"Inject" @@ -74,19 +71,13 @@ module Request (L1_operation : INJECTOR_OPERATION) = struct match r with | Add_pending op -> Format.fprintf ppf "request add %a to pending queue" L1_operation.pp op - | New_tezos_head ((block, level), r) -> + | New_tezos_head (block, level) -> Format.fprintf ppf "switching to new Tezos head %a at level %ld" Block_hash.pp block - level ; - if r.old_chain <> [] || r.new_chain <> [] then - Format.fprintf - ppf - ", with reorg of -%d +%d" - (List.length r.old_chain) - (List.length r.new_chain) + level | Inject -> Format.fprintf ppf "injection" end diff --git a/src/lib_injector/injector_worker_types.mli b/src/lib_injector/injector_worker_types.mli index a9a227828710..829bf8037613 100644 --- a/src/lib_injector/injector_worker_types.mli +++ b/src/lib_injector/injector_worker_types.mli @@ -1,7 +1,8 @@ (*****************************************************************************) (* *) (* Open Source License *) -(* Copyright (c) 2022 Nomadic Labs, *) +(* Copyright (c) 2023 Nomadic Labs, *) +(* Copyright (c) 2023 Functori, *) (* *) (* Permission is hereby granted, free of charge, to any person obtaining a *) (* copy of this software and associated documentation files (the "Software"),*) @@ -28,9 +29,7 @@ open Injector_sigs module Request (Inj_operation : INJECTOR_OPERATION) : sig type ('a, 'b) t = | Add_pending : Inj_operation.t -> (unit, error trace) t - | New_tezos_head : - (Block_hash.t * int32) * (Block_hash.t * int32) Reorg.t - -> (unit, error trace) t + | New_tezos_head : (Block_hash.t * int32) -> (unit, error trace) t | Inject : (unit, error trace) t type view = View : _ t -> view diff --git a/src/proto_016_PtMumbai/lib_sc_rollup_node/daemon.ml b/src/proto_016_PtMumbai/lib_sc_rollup_node/daemon.ml index 22c27b36840e..7092774bce71 100644 --- a/src/proto_016_PtMumbai/lib_sc_rollup_node/daemon.ml +++ b/src/proto_016_PtMumbai/lib_sc_rollup_node/daemon.ml @@ -372,22 +372,6 @@ module Make (PVM : Pvm.S) = struct in return_unit - let notify_injector new_head (reorg : Layer1.head Reorg.t) = - let open Lwt_result_syntax in - let open Layer1 in - let new_chain = - List.map (fun {hash; level} -> (hash, level)) reorg.new_chain - in - let old_chain = - List.map (fun {hash; level} -> (hash, level)) reorg.old_chain - in - let*! () = - Injector.new_tezos_head - (new_head.hash, new_head.level) - {new_chain; old_chain} - in - return_unit - (* [on_layer_1_head node_ctxt head] processes a new head from the L1. It also processes any missing blocks that were not processed. Every time a head is processed we also process head~2 as finalized (which may recursively @@ -431,7 +415,6 @@ module Make (PVM : Pvm.S) = struct let* () = List.iter_es (process_head node_ctxt) reorg.new_chain in let* () = Components.Commitment.Publisher.publish_commitments () in let* () = Components.Commitment.Publisher.cement_commitments () in - let* () = notify_injector head reorg in let*! () = Daemon_event.new_heads_processed reorg.new_chain in let* () = Components.Refutation_game.process head node_ctxt in let* () = Components.Batcher.batch () in diff --git a/src/proto_alpha/lib_sc_rollup_node/daemon.ml b/src/proto_alpha/lib_sc_rollup_node/daemon.ml index 312f6832b212..2551d538fa7f 100644 --- a/src/proto_alpha/lib_sc_rollup_node/daemon.ml +++ b/src/proto_alpha/lib_sc_rollup_node/daemon.ml @@ -369,22 +369,6 @@ module Make (PVM : Pvm.S) = struct in return_unit - let notify_injector new_head (reorg : Layer1.head Reorg.t) = - let open Lwt_result_syntax in - let open Layer1 in - let new_chain = - List.map (fun {hash; level} -> (hash, level)) reorg.new_chain - in - let old_chain = - List.map (fun {hash; level} -> (hash, level)) reorg.old_chain - in - let*! () = - Injector.new_tezos_head - (new_head.hash, new_head.level) - {new_chain; old_chain} - in - return_unit - (* [on_layer_1_head node_ctxt head] processes a new head from the L1. It also processes any missing blocks that were not processed. Every time a head is processed we also process head~2 as finalized (which may recursively @@ -428,7 +412,6 @@ module Make (PVM : Pvm.S) = struct let* () = List.iter_es (process_head node_ctxt) reorg.new_chain in let* () = Components.Commitment.Publisher.publish_commitments () in let* () = Components.Commitment.Publisher.cement_commitments () in - let* () = notify_injector head reorg in let*! () = Daemon_event.new_heads_processed reorg.new_chain in let* () = Components.Refutation_game.process head node_ctxt in let* () = Components.Batcher.batch () in -- GitLab