Add block processing queue to watcher info #1560

mederic-p · 2020-06-02T04:52:13Z

Overview

This PR adds a pending block queue to the watcher info.
Watcher info can take a lot of time to process blocks received so we store the block data as a big binary in a queue and then we process this queue one by one.
Queue length is regularly polled and sent to Datadog to check that we're not falling behind under the pending_block_queue_length metric name.

The monitor should be updated.

Changes

Add pending_blocks table
Add PendingBlockProcessor genserver to process pending blocks`
Add PendingBlockQueueLengthChecker genserver to poll queue length

Testing

Run tests.

apps/omg_watcher_info/test/support/db_case.ex

mix.lock

apps/omg_watcher_info/test/support/factories/pending_block_factory.ex

apps/omg_watcher_info/test/omg_watcher_info/pending_block_processor_test.exs

apps/omg_watcher_info/test/omg_watcher_info/db/pending_block_test.exs

apps/omg_watcher_info/lib/omg_watcher_info/supervisor.ex

apps/omg_watcher_info/lib/omg_watcher_info/pending_block_processor.ex

InoMurko · 2020-06-04T08:19:39Z

apps/omg_watcher_info/lib/omg_watcher_info/block_application_consumer.ex

-      |> to_mined_block()
-      |> DB.Block.insert_with_transactions()
+      |> to_pending_block()
+      |> DB.PendingBlock.insert()


PendingBlock is somewhat different naming then I would expect. the world queue is not used anywhere :)))

unnawut

what ino said

InoMurko · 2020-06-11T16:14:09Z

apps/omg_watcher/test/support/integration/test_helper.ex

+  state and add some "random" sleep to give the database time to process and write the block.
+  This function will instead poll for block.get until found (or timeout).
+  """
+  def wait_for_block_inserted_in_db(block_nr, timeout) do


here's my take on these helper functions. they're bloating the test suite and making the tests themselves harder to read. I prefer my tests self complete. Everything about a test is exactly in the test module.

InoMurko

couple of complaints, but good progress overall!

apps/omg_watcher/test/support/watcher_helper.ex

apps/omg_watcher_info/lib/omg_watcher_info/db/block.ex

apps/omg_watcher_info/lib/omg_watcher_info/db/pending_block.ex

apps/omg_watcher_info/lib/omg_watcher_info/pending_block_processor/storage.ex

InoMurko · 2020-06-11T16:22:51Z

apps/omg_watcher_info/lib/omg_watcher_info/pending_block_processor.ex

+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+defmodule OMG.WatcherInfo.PendingBlockProcessor do


all the OTP goodies. really clean.

InoMurko · 2020-06-11T16:23:42Z

apps/omg_watcher_info/priv/repo/migrations/20200529085008_create_pending_block_table.exs

+
+  def change() do
+    create table(:pending_blocks, primary_key: false) do
+      add :blknum, :bigint, null: false, primary_key: true


null false! 👍

InoMurko · 2020-06-11T16:28:02Z

apps/omg_watcher_info/test/omg_watcher_info/block_application_consumer_test.exs

+      alice = TestHelper.generate_entity()
+      bob = TestHelper.generate_entity()
+
+      tx_1 = TestHelper.create_recovered([{1, 0, 0, alice}], @eth, [{bob, 300}])
+      tx_2 = TestHelper.create_recovered([{1, 0, 0, alice}], @eth, [{bob, 500}])
+


are these perhaps unnecessarily complex? the data that block_application_consumer can accept is really arbitrary. and the test should only prove that :term_to_binary and :binary_to_term does what it needs to do

InoMurko · 2020-06-11T16:28:40Z

apps/omg_watcher_info/test/omg_watcher_info/block_application_consumer_test.exs

+    Process.send(pid, {:internal_event_bus, :block_received, block}, [:noconnect])
+
+    # this waits for all messages in process inbox is processed
+    _ = :sys.get_state(pid)


InoMurko · 2020-06-11T16:42:54Z

apps/omg_watcher_info/test/omg_watcher_info/pending_block_processor_test.exs

+    end
+
+    test "calls get_next_pending_block/0 when triggered", %{pid: pid} do
+      storage_pid = start_storage_mock()


hm... I don't understand how this works.
you're calling start_storage_mock/0 in every test, but this process is a named process:
GenServer.start_link(__MODULE__, args, name: __MODULE__).

The internal state of this module is being updated and accessed from all tests (except the ones below). So the only explanation is that these are not executed async?

okay, so the reason is that everything inside a describe is executed top-down.
which wasn't obvious to me.
here's a couple of suggestions:

move the DumbStorage to the bottom

move setup_with_storage/0 to the bottom

couple DumpStorage with the instance of PendingBlockProcessor with an Agent. Agent holds key pairs %{pending block processor pid => dump storage pid}. then you can drop the DumpStorage naming.

To me, it's easier to reason about your tests. I didn't understand why all tests start new PendingBlockProcessor but they all funnel towards the same instance of storage.

Also, it seems you could have accomplished the same thing with Postgres Sandbox?

Will remove the mocked storage, the genserver is too simple now to require mocking.

apps/omg_watcher/test/support/integration/test_helper.ex

InoMurko · 2020-06-12T17:57:10Z

apps/omg_watcher_info/lib/omg_watcher_info/configuration.ex

+    Application.fetch_env!(@app, :pending_block_processing_interval)
+  end
+
+  def pending_block_queue_length_check_interval() do


should we find a compromise in naming?
would block_queue_check_interval suffice?
similarly above. The word queue implies pending. And queue is a lot more common naming standard.

InoMurko · 2020-06-18T16:51:49Z

apps/omg_watcher/test/support/integration/test_helper.ex

    # TODO query to State used in tests instead of an event system, remove when event system is here
    fn ->
-      if State.get_status() |> elem(0) <= block_nr,
+      if State.get_status() |> elem(0) <= block_number,


Not sure what this is, but we need to be careful in comparisons.

number < atom < reference < fun < port < pid < tuple < map < nil < list < bit string

InoMurko · 2020-06-18T16:55:23Z

config/config.exs

+  pending_block_processing_interval: 1000,
+  block_queue_check_interval: 10_000


There's no particular reason why these two constants live in a configuration, right? They're not reused or available to configuration from outside. They could live in the module itself.

It helps with seeing an overview of all the rhythms in the system though!

…nfo-block-processing

unnawut

lgtm, approving with some minor comments

unnawut · 2020-06-23T07:55:01Z

apps/omg_watcher_info/lib/omg_watcher_info/application.ex

+
+  def start_phase(:attach_telemetry, :normal, _phase_args) do
+    handlers = [
+      ["measure-global", OMG.WatcherInfo.Measure.supported_events(), &OMG.WatcherInfo.Measure.handle_event/4, nil]


Suggested change

["measure-global", OMG.WatcherInfo.Measure.supported_events(), &OMG.WatcherInfo.Measure.handle_event/4, nil]

["measure-watcher-info", OMG.WatcherInfo.Measure.supported_events(), &OMG.WatcherInfo.Measure.handle_event/4, nil]

to follow the existing pattern:

"measure-state" -> OMG.State.Measure.*

"measure-db" => OMG.DB.Measure.*

unnawut · 2020-06-23T07:56:34Z

apps/omg_watcher/test/support/integration/test_helper.ex

    # TODO query to State used in tests instead of an event system, remove when event system is here
    fn ->
-      if State.get_status() |> elem(0) <= block_nr,
+      if State.get_status() |> elem(0) <= block_number,


Suggested change

if State.get_status() |> elem(0) <= block_number,

if elem(State.get_status(), 0) <= block_number,

unnawut · 2020-06-23T08:07:34Z

apps/omg_watcher_info/lib/omg_watcher_info/db/block.ex

@@ -170,6 +174,7 @@ defmodule OMG.WatcherInfo.DB.Block do
      |> prepare_inserts(db_txs_stream, "db_txs_", DB.Transaction)
      |> prepare_inserts(db_outputs_stream, "db_outputs_", DB.TxOutput)
      |> DB.TxOutput.spend_utxos(db_inputs)
+      |> Ecto.Multi.delete("pending_block", pending_block, [])


:mindblown: Interesting approach

unnawut · 2020-06-23T08:58:57Z

apps/omg_watcher_info/lib/omg_watcher_info/db/pending_block.ex

+    (pending_block in __MODULE__)
+    |> from(order_by: :blknum, limit: 1)
+    |> Repo.all()
+    |> Enum.at(0)


(pending_block in __MODULE__) |> from(order_by: :blknum, limit: 1) |> Repo.one()

a bit less implementation to worry about

you can also remove the limit: 1 ;)

I can go with Repo.one but need to keep the limit: 1 otherwise it'll raise when there is more than 1 pending block in the table, https://hexdocs.pm/ecto/Ecto.Repo.html#c:one/2 :)

unnawut · 2020-06-23T16:53:13Z

apps/omg_watcher_info/lib/omg_watcher_info/db/block.ex

-      }) do
+  # sobelow_skip ["Misc.BinToTerm"]
+  @spec insert_pending_block(PendingBlock.t()) :: {:ok, %__MODULE__{}} | {:error, any()}
+  def insert_pending_block(pending_block) do


Not sure if this is just me. I got confused whether it's inserting a new pending block vs. inserting an existing pending block into a proper block. Especially when reading from its caller, e.g:

elixir-omg/apps/omg_watcher_info/lib/omg_watcher_info/pending_block_processor/storage.ex

Lines 29 to 31 in 0ee6d37

def process_block(block) do

Block.insert_pending_block(block)

end

and when there are similar calls next to each other:

elixir-omg/apps/omg_watcher_info/test/omg_watcher_info/db/block_test.exs

Lines 334 to 336 in 0ee6d37

pending_block = insert(:pending_block, %{data: :erlang.term_to_binary(mined_block), blknum: 1000})

{:ok, block} = DB.Block.insert_pending_block(pending_block)

Can we name it insert_from_pending_block() instead?

Correct, will update 👍

unnawut · 2020-06-23T17:04:02Z

apps/omg_watcher_info/test/omg_watcher_info/pending_block_processor_test.exs

+      :erlang.trace(pid, true, [:receive])
+
+      assert get_all_pending_blocks() == []
+      assert get_all_blocks() == []
+
+      %{blknum: blknum_1} = insert(:pending_block)
+      %{blknum: blknum_2} = insert(:pending_block)
+      %{blknum: blknum_3} = insert(:pending_block)
+
+      assert_receive {:trace, ^pid, :receive, :timeout}, @interval + 1
+      get_state(pid)
+
+      assert [%{blknum: ^blknum_1}] = get_all_blocks()
+      assert [%{blknum: ^blknum_2}, %{blknum: ^blknum_3}] = get_all_pending_blocks()
+
+      assert_receive {:trace, ^pid, :receive, :timeout}
+      get_state(pid)
+
+      assert [%{blknum: ^blknum_3}] = get_all_pending_blocks()
+      assert [%{blknum: ^blknum_1}, %{blknum: ^blknum_2}] = get_all_blocks()
+
+      assert_receive {:trace, ^pid, :receive, :timeout}
+      get_state(pid)


Oh interesting use of trace & assert

Thanks Ino for the suggestion!

unnawut · 2020-06-23T17:05:37Z

apps/omg_watcher_info/test/omg_watcher_info/pending_block_processor_test.exs

+      %{blknum: blknum_2} = insert(:pending_block)
+      %{blknum: blknum_3} = insert(:pending_block)
+
+      assert_receive {:trace, ^pid, :receive, :timeout}, @interval + 1


What does + 1 represent? I'm assuming it's for the DB insert but not too sure and will be afraid to change.

Maybe name it in a variable or add as comment to help future maintainers?

It's actually not needed, just need to wait for at least @inteval, will remove.

unnawut · 2020-06-23T17:08:21Z

apps/omg_watcher_info/test/omg_watcher_info/pending_block_queue_length_checker_test.exs

+    handler_id = {__MODULE__, :rand.uniform(100)}
+
+    on_exit(fn ->
+      :telemetry.detach(handler_id)
+    end)
+
+    Map.put(tags, :handler_id, handler_id)


nice nice nice

unnawut · 2020-06-23T17:12:21Z

config/config.exs

+  pending_block_processing_interval: 1000,
+  block_queue_check_interval: 10_000


It helps with seeing an overview of all the rhythms in the system though!

* feat: sync v1.0.1 changes back to master (#1599) * Inomurko/reorg block getter (#1554) * dont store blockgetter events * dont fetch blockgetter events through aggregator * get_block_submitted_events * get_block_submitted_events * return tuple * the right contract encoding * unused api function * prevent race condition for status cache (#1558) * prevent race condition for status cache * Changelog for v1.0.0 (#1556) * restart strategy (#1565) * restart strategy * restart strategy * restart strategy * Update changelog for v1.0.0 * global block get interval (#1576) * Update changelog for v1.0.0 * feat: increase ExitProcessor timeouts (#1592) * increase timeouts * docs: changelog Co-authored-by: Unnawut Leepaisalsuwanna <unnawut@omisego.co> * chore: update watcher docker-compose to v1.0.1 * docs: small non-content fix to changelog Co-authored-by: Ino Murko <ino.murko@outlook.com> Co-authored-by: Thibault <thibault@omisego.co> * Add block processing queue to watcher info (#1560) * [WIP] initial queuing work * refactor: queue processing * catch db timeouts * update existing tests * add tests * continue with tests * finish tests * Fix integration test * refactor * refactor * refactor tests * remove retry_count * remove on exist genserver shutdown * add telemetry queue length event * sobelow skip BinToTerm * remove pending block status * naming * fix tests * fix tests * rename config * remove unused alias * fix PR minor comments * missing file * Add Transaction filter by end_datetime (#1595) * add new query * add if case * format * refactor * add test * fix feature spec * use address * fix assert * fix test * dummy assert * add get single tx * add wait tx * working waiter pooler * add wait helper cabbage test * test end_datetime querty * add swagger generator spec * use second * fix allow constraints * update constraints validator * fix test * fix test * edit name * add empty line' * more fixes * fix warning * refactor use the elixir way * improve test * fix test and typo * use cheap fn * sort * use any * use any boolean return * use Enum.all * final credo * format * use default params * update constaints test for end_datetime * mix format * @default_paging constant * fix test end_datetime * fix test and rename feature file' * mix format * Revert "explain analyze updates (#1569)" (#1601) This reverts commit 3431f26. This commit is intended to be only deployed to develop env, thus a short life commit. Reverting this in preparation of release. * release artifacts (#1597) * release artifacts Co-authored-by: Unnawut Leepaisalsuwanna <921194+unnawut@users.noreply.github.com> Co-authored-by: Ino Murko <ino.murko@outlook.com> Co-authored-by: Thibault <thibault@omisego.co> Co-authored-by: Mederic <32560642+mederic-p@users.noreply.github.com> Co-authored-by: Jarindr Thitadilaka <jarindr23@gmail.com>

* chore: merge master back to v1.0.2 (#1606) * feat: sync v1.0.1 changes back to master (#1599) * Inomurko/reorg block getter (#1554) * dont store blockgetter events * dont fetch blockgetter events through aggregator * get_block_submitted_events * get_block_submitted_events * return tuple * the right contract encoding * unused api function * prevent race condition for status cache (#1558) * prevent race condition for status cache * Changelog for v1.0.0 (#1556) * restart strategy (#1565) * restart strategy * restart strategy * restart strategy * Update changelog for v1.0.0 * global block get interval (#1576) * Update changelog for v1.0.0 * feat: increase ExitProcessor timeouts (#1592) * increase timeouts * docs: changelog Co-authored-by: Unnawut Leepaisalsuwanna <unnawut@omisego.co> * chore: update watcher docker-compose to v1.0.1 * docs: small non-content fix to changelog Co-authored-by: Ino Murko <ino.murko@outlook.com> Co-authored-by: Thibault <thibault@omisego.co> * Add block processing queue to watcher info (#1560) * [WIP] initial queuing work * refactor: queue processing * catch db timeouts * update existing tests * add tests * continue with tests * finish tests * Fix integration test * refactor * refactor * refactor tests * remove retry_count * remove on exist genserver shutdown * add telemetry queue length event * sobelow skip BinToTerm * remove pending block status * naming * fix tests * fix tests * rename config * remove unused alias * fix PR minor comments * missing file * Add Transaction filter by end_datetime (#1595) * add new query * add if case * format * refactor * add test * fix feature spec * use address * fix assert * fix test * dummy assert * add get single tx * add wait tx * working waiter pooler * add wait helper cabbage test * test end_datetime querty * add swagger generator spec * use second * fix allow constraints * update constraints validator * fix test * fix test * edit name * add empty line' * more fixes * fix warning * refactor use the elixir way * improve test * fix test and typo * use cheap fn * sort * use any * use any boolean return * use Enum.all * final credo * format * use default params * update constaints test for end_datetime * mix format * @default_paging constant * fix test end_datetime * fix test and rename feature file' * mix format * Revert "explain analyze updates (#1569)" (#1601) This reverts commit 3431f26. This commit is intended to be only deployed to develop env, thus a short life commit. Reverting this in preparation of release. * release artifacts (#1597) * release artifacts Co-authored-by: Unnawut Leepaisalsuwanna <921194+unnawut@users.noreply.github.com> Co-authored-by: Ino Murko <ino.murko@outlook.com> Co-authored-by: Thibault <thibault@omisego.co> Co-authored-by: Mederic <32560642+mederic-p@users.noreply.github.com> Co-authored-by: Jarindr Thitadilaka <jarindr23@gmail.com> * docs: v1.0.2 change logs (#1611) * chore: bump version in VERSION file (#1613) Co-authored-by: Unnawut Leepaisalsuwanna <921194+unnawut@users.noreply.github.com> Co-authored-by: Ino Murko <ino.murko@outlook.com> Co-authored-by: Thibault <thibault@omisego.co> Co-authored-by: Mederic <32560642+mederic-p@users.noreply.github.com> Co-authored-by: Jarindr Thitadilaka <jarindr23@gmail.com>

unnawut reviewed Jun 4, 2020

View reviewed changes

apps/omg_watcher_info/test/support/db_case.ex Outdated Show resolved Hide resolved

InoMurko suggested changes Jun 4, 2020

View reviewed changes

Mederic added 8 commits June 4, 2020 15:51

[WIP] initial queuing work

968ecb2

refactor: queue processing

2ec0263

catch db timeouts

cc8950f

update existing tests

3abe795

add tests

8df1604

continue with tests

a0f1960

finish tests

bdd9af5

Fix integration test

8e0835e

unnawut reviewed Jun 4, 2020

View reviewed changes

mederic-p force-pushed the mederic-p/watcher-info-block-processing branch from 2d07a51 to 8e0835e Compare June 4, 2020 09:05

Mederic added 9 commits June 4, 2020 18:54

refactor

dbe8594

refactor

5c0d14b

Merge branch 'master' into mederic-p/watcher-info-block-processing

3e2bee1

refactor tests

3476c71

remove retry_count

6293606

remove on exist genserver shutdown

8df12f2

add telemetry queue length event

20b7404

Merge branch 'master' into mederic-p/watcher-info-block-processing

93b4d05

sobelow skip BinToTerm

a602b73

mederic-p marked this pull request as ready for review June 10, 2020 04:57

mederic-p requested review from achiurizo, T-Dnzt, InoMurko and unnawut June 10, 2020 04:57

InoMurko reviewed Jun 11, 2020

View reviewed changes

InoMurko suggested changes Jun 11, 2020

View reviewed changes

InoMurko reviewed Jun 11, 2020

View reviewed changes

apps/omg_watcher/test/support/integration/test_helper.ex Outdated Show resolved Hide resolved

remove pending block status

098be67

Mederic added 4 commits June 12, 2020 14:55

Merge branch 'master' into mederic-p/watcher-info-block-processing

7f85dea

naming

0cecf68

fix tests

8c337cd

fix tests

aee4bf7

InoMurko reviewed Jun 12, 2020

View reviewed changes

Mederic added 2 commits June 15, 2020 07:32

rename config

1813caf

remove unused alias

0ee6d37

mederic-p requested a review from InoMurko June 16, 2020 03:23

InoMurko reviewed Jun 18, 2020

View reviewed changes

InoMurko approved these changes Jun 18, 2020

View reviewed changes

mederic-p requested a review from pnowosie June 19, 2020 06:48

Merge remote-tracking branch 'origin/master' into mederic-p/watcher-i…

28c95fe

…nfo-block-processing

unnawut reviewed Jun 23, 2020

View reviewed changes

Mederic added 2 commits June 24, 2020 10:15

fix PR minor comments

1a299a3

missing file

277e574

unnawut approved these changes Jun 24, 2020

View reviewed changes

mederic-p merged commit 33f228c into master Jun 24, 2020

mederic-p deleted the mederic-p/watcher-info-block-processing branch June 24, 2020 04:51

unnawut added the enhancement New feature or request label Aug 28, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add block processing queue to watcher info #1560

Add block processing queue to watcher info #1560

mederic-p commented Jun 2, 2020 •

edited

Loading

InoMurko Jun 4, 2020

unnawut left a comment

InoMurko Jun 11, 2020

InoMurko left a comment

InoMurko Jun 11, 2020

InoMurko Jun 11, 2020

InoMurko Jun 11, 2020

InoMurko Jun 11, 2020

InoMurko Jun 11, 2020

InoMurko Jun 11, 2020

mederic-p Jun 12, 2020

InoMurko Jun 12, 2020

InoMurko Jun 18, 2020

InoMurko Jun 18, 2020

unnawut Jun 23, 2020

unnawut left a comment

unnawut Jun 23, 2020

unnawut Jun 23, 2020

unnawut Jun 23, 2020

unnawut Jun 23, 2020

InoMurko Jun 23, 2020

mederic-p Jun 24, 2020

unnawut Jun 23, 2020

mederic-p Jun 24, 2020

unnawut Jun 23, 2020

mederic-p Jun 24, 2020

unnawut Jun 23, 2020

mederic-p Jun 24, 2020

unnawut Jun 23, 2020

unnawut Jun 23, 2020

		pending_block_processing_interval: 1000,
		block_queue_check_interval: 10_000

	["measure-global", OMG.WatcherInfo.Measure.supported_events(), &OMG.WatcherInfo.Measure.handle_event/4, nil]
	["measure-watcher-info", OMG.WatcherInfo.Measure.supported_events(), &OMG.WatcherInfo.Measure.handle_event/4, nil]

	if State.get_status() \|> elem(0) <= block_number,
	if elem(State.get_status(), 0) <= block_number,

	def process_block(block) do
	Block.insert_pending_block(block)
	end

	pending_block = insert(:pending_block, %{data: :erlang.term_to_binary(mined_block), blknum: 1000})

	{:ok, block} = DB.Block.insert_pending_block(pending_block)

Add block processing queue to watcher info #1560

Add block processing queue to watcher info #1560

Conversation

mederic-p commented Jun 2, 2020 • edited Loading

Overview

Changes

Testing

Choose a reason for hiding this comment

unnawut left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

InoMurko left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

unnawut left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mederic-p commented Jun 2, 2020 •

edited

Loading