fix(watcher): do not download more blocks if last applied block is la…

…gging

apps/omg_watcher/config/config.exs

@@ -12,7 +12,8 @@ config :omg_watcher,
   ecto_repos: [OMG.Watcher.DB.Repo],
   slow_exit_validator_block_margin: 10,
   maximum_block_withholding_time_ms: 10_000,
-  block_getter_height_sync_interval_ms: 2_000
+  block_getter_height_sync_interval_ms: 2_000,
+  maximum_last_applied_block_lag: 20_000
 
 # Configures the endpoint
 config :omg_watcher, OMG.Watcher.Web.Endpoint,

apps/omg_watcher/lib/block_getter.ex

@@ -69,6 +69,7 @@ defmodule OMG.Watcher.BlockGetter do
       {state, synced_height, db_updates} = Core.apply_block(state, blknum, block_rootchain_height)
       :ok = RootChainCoordinator.check_in(synced_height, :block_getter)
       :ok = OMG.DB.multi_update(db_updates)
+      :producer = send(self(), :producer)
 
       {:noreply, state}
     else
@@ -100,14 +101,16 @@ defmodule OMG.Watcher.BlockGetter do
     :producer = send(self(), :producer)
 
     maximum_block_withholding_time_ms = Application.get_env(:omg_watcher, :maximum_block_withholding_time_ms)
+    maximum_last_applied_block_lag = Application.get_env(:omg_watcher, :maximum_last_applied_block_lag)
 
     {
       :ok,
       Core.init(
         child_top_block_number,
         child_block_interval,
         exact_synced_height,
-        maximum_block_withholding_time_ms: maximum_block_withholding_time_ms
+        maximum_block_withholding_time_ms: maximum_block_withholding_time_ms,
+        maximum_last_applied_block_lag: maximum_last_applied_block_lag
       )
     }
   end

apps/omg_watcher/lib/block_getter/core.ex

@@ -22,6 +22,8 @@ defmodule OMG.Watcher.BlockGetter.Core do
 
   use OMG.API.LoggerExt
 
+  @default_applied_block_lag 20 * 1_000
+
   defmodule PotentialWithholding do
     @moduledoc false
 
@@ -43,7 +45,8 @@ defmodule OMG.Watcher.BlockGetter.Core do
     :maximum_number_of_pending_blocks,
     :unapplied_blocks,
     :potential_block_withholdings,
-    :maximum_block_withholding_time_ms
+    :maximum_block_withholding_time_ms,
+    :maximum_last_applied_block_lag
   ]
 
   @type t() :: %__MODULE__{
@@ -60,7 +63,8 @@ defmodule OMG.Watcher.BlockGetter.Core do
           potential_block_withholdings: %{
             non_neg_integer => pos_integer
           },
-          maximum_block_withholding_time_ms: pos_integer
+          maximum_block_withholding_time_ms: pos_integer,
+          maximum_last_applied_block_lag: pos_integer()
         }
 
   @type block_error() ::
@@ -95,7 +99,8 @@ defmodule OMG.Watcher.BlockGetter.Core do
       maximum_number_of_pending_blocks: Keyword.get(opts, :maximum_number_of_pending_blocks, 10),
       unapplied_blocks: %{},
       potential_block_withholdings: %{},
-      maximum_block_withholding_time_ms: Keyword.get(opts, :maximum_block_withholding_time_ms, 0)
+      maximum_block_withholding_time_ms: Keyword.get(opts, :maximum_block_withholding_time_ms, 0),
+      maximum_last_applied_block_lag: Keyword.get(opts, :maximum_last_applied_block_lag, @default_applied_block_lag)
     }
   end
 
@@ -200,9 +205,11 @@ defmodule OMG.Watcher.BlockGetter.Core do
         %__MODULE__{
           num_of_heighest_block_being_downloaded: num_of_heighest_block_being_downloaded,
           block_interval: block_interval,
+          last_applied_block: last_applied_block,
           number_of_blocks_being_downloaded: number_of_blocks_being_downloaded,
           potential_block_withholdings: potential_block_withholdings,
-          maximum_number_of_pending_blocks: maximum_number_of_pending_blocks
+          maximum_number_of_pending_blocks: maximum_number_of_pending_blocks,
+          maximum_last_applied_block_lag: maximum_last_applied_block_lag
         } = state,
         next_child
       ) do
@@ -215,7 +222,7 @@ defmodule OMG.Watcher.BlockGetter.Core do
     potential_next_block_numbers =
       first_block_number
       |> Stream.iterate(&(&1 + block_interval))
-      |> Stream.take_while(&(&1 < next_child))
+      |> Stream.take_while(&(&1 < next_child and &1 <= last_applied_block + maximum_last_applied_block_lag))
       |> Enum.to_list()
 
     blocks_numbers =

apps/omg_watcher/test/block_getter/core_test.exs

@@ -474,4 +474,26 @@ defmodule OMG.Watcher.BlockGetter.CoreTest do
         2
       )
   end
+
+  test "do not download blocks when last consumed blocks lags behind" do
+    interval = 1_000
+
+    {state, [1_000, 2_000, 3_000]} =
+      0
+      |> Core.init(interval, 0, maximum_number_of_pending_blocks: 5, maximum_last_applied_block_lag: 3000)
+      |> Core.get_numbers_of_blocks_to_download(5_000)
+
+    state = handle_downloaded_block(state, %Block{number: 1_000})
+    synced_height = 1
+
+    {_, _, _, state} =
+      Core.get_blocks_to_apply(
+        state,
+        [%{blknum: 1_000, eth_height: synced_height}],
+        synced_height
+      )
+
+    {state, _, _} = Core.apply_block(state, 1_000, synced_height)
+    {_, [4_000]} = Core.get_numbers_of_blocks_to_download(state, 5_000)
+  end
 end