Using CRDTs for the processes only with Horde

Author: jyeshe

lib/lightning/application.ex

@@ -105,6 +105,12 @@ defmodule Lightning.Application do
       [
         Lightning.PromEx,
         {Cluster.Supervisor, [topologies, [name: Lightning.ClusterSupervisor]]},
+        {Horde.Registry,
+         name: Lightning.HordeRegistry, keys: :unique, members: :auto},
+        {Horde.DynamicSupervisor,
+         name: Lightning.DistributedSupervisor,
+         strategy: :one_for_one,
+         members: :auto},
         {Lightning.Vault, Application.get_env(:lightning, Lightning.Vault, [])},
         # Start the Ecto repository
         Lightning.Repo,
@@ -174,6 +180,15 @@ defmodule Lightning.Application do
     :ok
   end
 
+  def start_phase(:init_rate_limiter, :normal, _args) do
+    Horde.DynamicSupervisor.start_child(
+      Lightning.DistributedSupervisor,
+      Lightning.WebhookRateLimiter
+    )
+
+    :ok
+  end
+
   def oban_opts do
     opts = Application.get_env(:lightning, Oban)
 

lib/lightning/rate_limiters.ex

@@ -22,14 +22,6 @@ defmodule Lightning.RateLimiters do
             | {:deny, non_neg_integer()}
   end
 
-  defmodule Webhook do
-    @moduledoc false
-
-    use ReplicatedRateLimiter,
-      default_capacity: 10,
-      default_refill: 2
-  end
-
   @spec hit({:failure_email, String.t(), String.t()}) :: Mail.hit_result()
   def hit({:failure_email, workflow_id, user_id}) do
     [time_scale: time_scale, rate_limit: rate_limit] =
@@ -45,15 +37,6 @@ defmodule Lightning.RateLimiters do
     end
   end
 
-  def hit({:webhook, project_id}) do
-    # 10 requests for a second, then 2 requests per second
-    # Over a long enough period of time, this will allow 2 requests per second.
-    # allow?("webhook_#{project_id}", 10, 2)
-    # capacity and refill is by design a module attribute
-    # TODO: passing it here might eliminate the need for macro for easier maintainance
-    Webhook.allow?("webhook_#{project_id}")
-  end
-
   def child_spec(opts) do
     %{
       id: __MODULE__,
@@ -63,7 +46,7 @@ defmodule Lightning.RateLimiters do
   end
 
   def start_link(opts) do
-    children = [{Mail, opts}, {Webhook, opts}]
+    children = [{Mail, opts}]
     Supervisor.start_link(children, strategy: :one_for_one)
   end
 end

lib/lightning/webhook_rate_limiter.ex

@@ -0,0 +1,98 @@
+defmodule Lightning.WebhookRateLimiter do
+  @moduledoc false
+  use GenServer
+
+  @capacity 10
+  @refill_per_sec 2
+
+  require Logger
+
+  def child_spec(opts) do
+    {id, name} =
+      if name = Keyword.get(opts, :name) do
+        {"#{__MODULE__}_#{name}", name}
+      else
+        {__MODULE__, __MODULE__}
+      end
+
+    %{
+      id: id,
+      start: {__MODULE__, :start_link, [name]},
+      shutdown: 10_000,
+      restart: :transient
+    }
+  end
+
+  def start_link(name) do
+    with {:error, {:already_started, pid}} <-
+           GenServer.start_link(__MODULE__, [], name: via_tuple(name)) do
+      Logger.info("already started at #{inspect(pid)}, returning :ignore")
+      :ignore
+    end
+  end
+
+  @impl true
+  def init([]) do
+    Process.flag(:trap_exit, true)
+
+    {:ok, %{table: :ets.new(:table, [:set])}}
+  end
+
+  def check_rate(bucket, cost \\ 1, name \\ __MODULE__) do
+    name
+    |> via_tuple()
+    |> GenServer.call({:check_rate, bucket, cost})
+  end
+
+  def inspect_table(name \\ __MODULE__),
+    do: GenServer.call(via_tuple(name), :inspect_table)
+
+  @impl true
+  def handle_call({:check_rate, bucket, cost}, _from, %{table: table} = state) do
+    {:reply, do_check_rate(table, bucket, cost), state}
+  end
+
+  @impl true
+  def handle_call(:inspect_table, _from, %{table: table} = state) do
+    {:reply, :ets.info(table), state}
+  end
+
+  @impl true
+  def handle_info(
+        {:EXIT, _from, {:name_conflict, {_key, _value}, registry, pid}},
+        state
+      ) do
+    Logger.info(
+      "Stopping #{inspect({registry, pid})} as it has already started in another node."
+    )
+
+    {:stop, :normal, state}
+  end
+
+  def do_check_rate(table, bucket, cost) do
+    now = System.monotonic_time(:millisecond)
+
+    :ets.insert_new(table, {bucket, {@capacity, now}})
+    [{^bucket, {level, updated}}] = :ets.lookup(table, bucket)
+
+    refilled = div((now - updated), 1_000) * @refill_per_sec
+    current = min(@capacity, level + refilled)
+
+    if current >= cost do
+      level = current - cost
+      :ets.insert(table, {bucket, {level, now}})
+
+      {:allow, level}
+    else
+      wait_ms = ceil((cost - current) / @refill_per_sec * 1_000) |> round()
+
+      {:deny, wait_ms}
+    end
+  end
+
+  def capacity, do: @capacity
+  def refill_per_second, do: @refill_per_sec
+
+  def via_tuple(name),
+    do: {:via, Horde.Registry, {Lightning.HordeRegistry, name}}
+end

mix.exs

@@ -55,7 +55,7 @@ defmodule Lightning.MixProject do
     [
       mod: {Lightning.Application, [:timex]},
       extra_applications: [:logger, :runtime_tools, :os_mon, :scrivener],
-      start_phases: [seed_prom_ex_telemetry: []]
+      start_phases: [seed_prom_ex_telemetry: [], init_rate_limiter: []]
     ]
   end
 
@@ -69,6 +69,7 @@ defmodule Lightning.MixProject do
   defp deps do
     [
       # {:rexbug, ">= 1.0.0", only: :test},
+      {:horde, "~> 0.9.0"},
       {:bcrypt_elixir, "~> 3.2"},
       {:bodyguard, "~> 2.2"},
       {:broadway_kafka, "~> 0.4.2"},

mix.lock

@@ -61,6 +61,7 @@
   "hammer": {:hex, :hammer, "7.0.1", "136edcd81af44becbe6b73a958c109e2364ab0dc026d7b19892037dc2632078c", [:mix], [], "hexpm", "796edf14ab2aa80df72080210fcf944ee5e8868d8ece7a7511264d802f58cc2d"},
   "hammer_backend_mnesia": {:hex, :hammer_backend_mnesia, "0.7.0", "b2a8cccc1d3506bc4cf8e95750fa5bd491390f227e9a0d981ad375291d7bd1dc", [:mix], [{:hammer, "~> 7.0", [hex: :hammer, repo: "hexpm", optional: false]}], "hexpm", "f77a3d54df865aa8137926df6fadac2a81c06f1c1c22f4e98e32392a22bf9e3e"},
   "heroicons": {:hex, :heroicons, "0.5.6", "95d730e7179c633df32d95c1fdaaecdf81b0da11010b89b737b843ac176a7eb5", [:mix], [{:castore, ">= 0.0.0", [hex: :castore, repo: "hexpm", optional: false]}, {:phoenix_live_view, ">= 0.18.2", [hex: :phoenix_live_view, repo: "hexpm", optional: false]}], "hexpm", "ca267f02a5fa695a4178a737b649fb6644a2e399639d4ba7964c18e8a58c2352"},
+  "horde": {:hex, :horde, "0.9.0", "522342bd7149aeed453c97692a8bca9cf7c9368c5a489afd802e575dc8df54a6", [:mix], [{:delta_crdt, "~> 0.6.2", [hex: :delta_crdt, repo: "hexpm", optional: false]}, {:libring, "~> 1.4", [hex: :libring, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}, {:telemetry_poller, "~> 0.5.0 or ~> 1.0", [hex: :telemetry_poller, repo: "hexpm", optional: false]}], "hexpm", "fae11e5bc9c980038607d0c3338cdf7f97124a5d5382fd4b6fb6beaab8e214fe"},
   "hpax": {:hex, :hpax, "1.0.2", "762df951b0c399ff67cc57c3995ec3cf46d696e41f0bba17da0518d94acd4aac", [:mix], [], "hexpm", "2f09b4c1074e0abd846747329eaa26d535be0eb3d189fa69d812bfb8bfefd32f"},
   "httpoison": {:hex, :httpoison, "2.2.1", "87b7ed6d95db0389f7df02779644171d7319d319178f6680438167d7b69b1f3d", [:mix], [{:hackney, "~> 1.17", [hex: :hackney, repo: "hexpm", optional: false]}], "hexpm", "51364e6d2f429d80e14fe4b5f8e39719cacd03eb3f9a9286e61e216feac2d2df"},
   "idna": {:hex, :idna, "6.1.1", "8a63070e9f7d0c62eb9d9fcb360a7de382448200fbbd1b106cc96d3d8099df8d", [:rebar3], [{:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "92376eb7894412ed19ac475e4a86f7b413c1b9fbb5bd16dccd57934157944cea"},
@@ -72,6 +73,7 @@
   "junit_formatter": {:hex, :junit_formatter, "3.4.0", "d0e8db6c34dab6d3c4154c3b46b21540db1109ae709d6cf99ba7e7a2ce4b1ac2", [:mix], [], "hexpm", "bb36e2ae83f1ced6ab931c4ce51dd3dbef1ef61bb4932412e173b0cfa259dacd"},
   "kafka_protocol": {:hex, :kafka_protocol, "4.1.9", "7c10d9adaba84c6f176f152e6ba8029c46dfb7cb12432587009128836cf9a44a", [:rebar3], [{:crc32cer, "0.1.11", [hex: :crc32cer, repo: "hexpm", optional: false]}], "hexpm", "14f89eed8329ff4c7b5448e318ee20a98bf5c1e5dc41b74b8af459dfb7590cef"},
   "libcluster": {:hex, :libcluster, "3.4.1", "271d2da892763bbef53c2872036c936fe8b80111eb1feefb2d30a3bb15c9b4f6", [:mix], [{:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.3", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "1d568157f069c6afa70ec0d736704cf799734bdbb6343f0322af4a980301c853"},
+  "libring": {:hex, :libring, "1.7.0", "4f245d2f1476cd7ed8f03740f6431acba815401e40299208c7f5c640e1883bda", [:mix], [], "hexpm", "070e3593cb572e04f2c8470dd0c119bc1817a7a0a7f88229f43cf0345268ec42"},
   "makeup": {:hex, :makeup, "1.1.2", "9ba8837913bdf757787e71c1581c21f9d2455f4dd04cfca785c70bbfff1a76a3", [:mix], [{:nimble_parsec, "~> 1.2.2 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "cce1566b81fbcbd21eca8ffe808f33b221f9eee2cbc7a1706fc3da9ff18e6cac"},
   "makeup_eex": {:hex, :makeup_eex, "0.1.2", "93a5ef3d28ed753215dba2d59cb40408b37cccb4a8205e53ef9b5319a992b700", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.16 or ~> 1.0", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_html, "~> 0.1.0 or ~> 1.0", [hex: :makeup_html, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "6140eafb28215ad7182282fd21d9aa6dcffbfbe0eb876283bc6b768a6c57b0c3"},
   "makeup_elixir": {:hex, :makeup_elixir, "0.16.2", "627e84b8e8bf22e60a2579dad15067c755531fea049ae26ef1020cad58fe9578", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "41193978704763f6bbe6cc2758b84909e62984c7752b3784bd3c218bb341706b"},

test/lightning/webhook_rate_limiter_test.exs

@@ -0,0 +1,100 @@
+defmodule Lightning.WebhookRateLimiterTest do
+  @moduledoc false
+  use ExUnit.Case
+
+  alias Lightning.WebhookRateLimiter
+
+  @default_capacity 10
+
+  describe "check_rate/2" do
+    test "allows up to the capacity and refills on multiple buckets" do
+      initial_capacity = @default_capacity
+      bucket1 = "project#{System.unique_integer()}"
+      bucket2 = "project#{System.unique_integer()}"
+
+      Enum.each(1..initial_capacity, fn i ->
+        level = initial_capacity - i
+        assert match?({:allow, ^level}, WebhookRateLimiter.check_rate(bucket1))
+        assert match?({:allow, ^level}, WebhookRateLimiter.check_rate(bucket2))
+      end)
+    end
+
+    test "denies after consuming the bucket" do
+      initial_capacity = @default_capacity
+      bucket1 = "project#{System.unique_integer()}"
+      bucket2 = "project#{System.unique_integer()}"
+
+      Enum.each(1..initial_capacity, fn i ->
+        assert {:allow, level} = WebhookRateLimiter.check_rate(bucket1)
+        assert level == initial_capacity - i
+      end)
+
+      assert {:allow, level} = WebhookRateLimiter.check_rate(bucket2)
+      assert level == initial_capacity - 1
+
+      assert {:deny, wait_ms} = WebhookRateLimiter.check_rate(bucket1)
+      assert 0 < wait_ms and wait_ms < 1_000
+    end
+
+    # Synthetic cluster not working.
+    # For testing use manual procedure:
+    # 0. Disable Endpoint server
+    # 1. Run node1 on one terminal: iex --sname node1@localhost --cookie hordecookie -S mix phx.server
+    # 2. Run node2 on another terminal: iex --sname node2@localhost --cookie hordecookie -S mix phx.server
+    # 3. Call Lightning.WebhookRateLimiter.inspect_table() on both iex and they show the same ets table process and node.
+    @tag skip: true
+    test "consumes the bucket remotely" do
+      {:ok, peer, _node1, node2} = start_nodes(:node1, :node2, ~c"localhost")
+
+      :rpc.call(node2, Application, :ensure_all_started, [:mix])
+      :rpc.call(node2, Application, :ensure_all_started, [:lightning])
+
+      # Copy current code paths to the peer node
+      :rpc.call(node2, :code, :add_paths, [:code.get_path()])
+
+      assert [
+               {Lightning.DistributedSupervisor, :node1@localhost},
+               {Lightning.DistributedSupervisor, :node2@localhost}
+             ] = Horde.Cluster.members(Lightning.DistributedSupervisor)
+
+      # initial_capacity = @default_capacity
+      bucket = "project#{System.unique_integer()}"
+
+      dbg(WebhookRateLimiter.check_rate(bucket))
+
+      # dbg :rpc.block_call(node1, WebhookRateLimiter, :inspect, [WebhookRateLimiter])
+      # dbg :rpc.block_call(node2, WebhookRateLimiter, :inspect, [WebhookRateLimiter])
+
+      # Enum.each(1..initial_capacity-1, fn i ->
+      #   assert {:allow, level} = :rpc.call(node2, WebhookRateLimiter, :check_rate, [bucket, 1])
+      #   assert level == initial_capacity - i - 1
+      # end)
+
+      # assert {:deny, wait_ms} = WebhookRateLimiter.check_rate(bucket)
+      # assert 0 < wait_ms and wait_ms < 1_000
+
+      :peer.stop(peer)
+    end
+  end
+
+  defp start_nodes(node1, node2, host) do
+    # Start the main node
+    node1_sname = :"#{node1}@#{host}"
+    {:ok, _pid} = Node.start(node1_sname, :shortnames)
+    true = Node.set_cookie(:delicious_cookie)
+    cookie = Node.get_cookie() |> to_charlist()
+
+    # Start the peer node
+    {:ok, peer, node2_sname} =
+      :peer.start(%{
+        name: node2,
+        host: host,
+        cookie: cookie,
+        args: [~c"-setcookie", cookie]
+      })
+
+    assert node2_sname in Node.list()
+
+    {:ok, peer, node1_sname, node2_sname}
+  end
+end

test/replicated_rate_limiter_test.exs

@@ -26,7 +26,9 @@ defmodule ReplicatedRateLimiterTest do
     Enum.each(1..9, fn _ -> CrdtEts.allow?("project2") end)
 
     assert_eventually(
-      DeltaCrdt.get(config.crdt_name, {"project2", "#{Node.self()}"}) |> dbg |> elem(0) == 0,
+      DeltaCrdt.get(config.crdt_name, {"project2", "#{Node.self()}"})
+      |> dbg
+      |> elem(0) == 0,
       1_000
     )
 
@@ -58,13 +60,12 @@ defmodule ReplicatedRateLimiterTest do
       DeltaCrdt.put(
         test_crdt,
         {"project2", "another_node"},
-        {10-i, System.system_time(:second)})
+        {10 - i, System.system_time(:second)}
+      )
     end)
 
     # Wait for the bucket to be updated
-    assert_eventually(
-      CrdtEts.to_list("project2") |> elem(0) == 1
-    )
+    assert_eventually(CrdtEts.to_list("project2") |> elem(0) == 1)
 
     assert {:allow, 4} = CrdtEts.allow?("project2", 10, 2)
   end

test/test_helper.exs

@@ -54,5 +54,12 @@ Application.put_env(:lightning, Lightning.Extensions,
   external_metrics: Lightning.Extensions.ExternalMetrics
 )
 
+epmd_path = System.find_executable("epmd")
+port = Port.open({:spawn_executable, epmd_path}, [])
+os_pid = Keyword.get(Port.info(port), :os_pid)
+
+# Configuring a "shutdown hook" to stop epmd after everything is done.
+System.at_exit(fn _ -> System.shell("kill -TERM #{os_pid}") end)
+
 ExUnit.start()
 Ecto.Adapters.SQL.Sandbox.mode(Lightning.Repo, :manual)