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/extensions/rate_limiting.ex

@@ -8,16 +8,9 @@ defmodule Lightning.Extensions.RateLimiting do
   @type message :: Lightning.Extensions.Message.t()
 
   defmodule Context do
-    @moduledoc """
-    Which user is making the request for a certain project.
-    """
+    @type t :: %Context{project_id: Ecto.UUID.t()}
 
-    @type t :: %Context{
-            project_id: Ecto.UUID.t(),
-            user_id: Ecto.UUID.t() | nil
-          }
-
-    defstruct [:project_id, :user_id]
+    defstruct [:project_id]
   end
 
   @callback limit_request(

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/services/rate_limiter.ex

@@ -7,7 +7,7 @@ defmodule Lightning.Services.RateLimiter do
   import Lightning.Services.AdapterHelper
 
   @impl true
-  def limit_request(conn, context, opts) do
+  def limit_request(conn, context, opts \\ []) do
     adapter().limit_request(conn, context, opts)
   end
 

lib/lightning/webhook_rate_limiter.ex

@@ -0,0 +1,100 @@
+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
+    name
+    |> via_tuple()
+    |> GenServer.call(:inspect_table)
+  end
+
+  @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
+      # can retry after 1 second
+      {:deny, 1}
+    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

lib/lightning_web/controllers/webhooks_controller.ex

@@ -1,7 +1,6 @@
 defmodule LightningWeb.WebhooksController do
   use LightningWeb, :controller
 
-  alias Lightning.Extensions.RateLimiting
   alias Lightning.Extensions.UsageLimiting.Action
   alias Lightning.Extensions.UsageLimiting.Context
   alias Lightning.Services.RateLimiter
@@ -10,7 +9,7 @@ defmodule LightningWeb.WebhooksController do
   alias Lightning.WorkOrders
 
   plug :reject_unfetched when action in [:create]
-  # plug :check_rate when action in [:create]
+  plug :check_rate when action in [:create]
 
   # Reject requests with unfetched body params, as they are not supported
   # See Plug.Parsers in Endpoint for more information.
@@ -28,20 +27,36 @@ defmodule LightningWeb.WebhooksController do
     end
   end
 
-  # defp check_rate(conn, _params) do
-  #   # TODO: this may be _after_ the body has been parsed (into body_params), so we may need to
-  #   # may need to move this plug further upstream.
-  #   case Lightning.RateLimiters.check_rate(conn.project_id) do
-  #     {:allow, _} ->
-  #       conn
-  #     {:deny, timeout} ->
-  #       conn
-  #       |> put_status(429)
-  #       |> put_resp_header("retry-after", to_string(timeout))
-  #       |> json(%{"error" => "Too many requests"})
-  #       |> halt()
-  #   end
-  # end
+  # Note: Plug.Parsers is called before the Router.
+  def check_rate(conn, _opts) do
+    with %Workflows.Trigger{enabled: true, workflow: %{project_id: project_id}} <-
+           conn.assigns.trigger,
+         :ok <- RateLimiter.limit_request(conn, %Context{project_id: project_id}) do
+      conn
+    else
+      %Workflows.Trigger{enabled: false} ->
+        conn
+        |> put_status(:forbidden)
+        |> json(%{
+          message:
+            "Unable to process request, trigger is disabled. Enable it on OpenFn to allow requests to this endpoint."
+        })
+        |> halt()
+
+      {:error, :too_many_requests, %{text: message}} ->
+        conn
+        |> put_status(:too_many_requests)
+        |> put_resp_header("retry-after", "1")
+        |> json(%{"error" => message})
+        |> halt()
+
+      _no_trigger_or_workflow ->
+        conn
+        |> put_status(:not_found)
+        |> json(%{"error" => "Webhook not found"})
+        |> halt()
+    end
+  end
 
   @spec check(Plug.Conn.t(), %{path: binary()}) :: Plug.Conn.t()
   def check(conn, _params) do
@@ -53,30 +68,25 @@ defmodule LightningWeb.WebhooksController do
   end
 
   @spec create(Plug.Conn.t(), %{path: binary()}) :: Plug.Conn.t()
-  def create(conn, _params) do
-    with %Workflows.Trigger{enabled: true, workflow: %{project_id: project_id}} =
-           trigger <- conn.assigns.trigger,
-         {:ok, without_run?} <- check_skip_run_creation(project_id),
-         :ok <-
-           RateLimiter.limit_request(
-             conn,
-             %RateLimiting.Context{project_id: project_id},
-             []
-           ) do
-      {:ok, work_order} =
-        WorkOrders.create_for(trigger,
-          workflow: trigger.workflow,
-          dataclip: %{
-            body: conn.body_params,
-            request: build_request(conn),
-            type: :http_request,
-            project_id: project_id
-          },
-          without_run: without_run?
-        )
-
-      conn |> json(%{work_order_id: work_order.id})
-    else
+  def create(%{assigns: %{trigger: trigger}} = conn, _params) do
+    %Workflows.Trigger{workflow: workflow} = trigger
+
+    case check_skip_run_creation(workflow.project_id) do
+      {:ok, without_run?} ->
+        {:ok, work_order} =
+          WorkOrders.create_for(trigger,
+            workflow: workflow,
+            dataclip: %{
+              body: conn.body_params,
+              request: build_request(conn),
+              type: :http_request,
+              project_id: workflow.project_id
+            },
+            without_run: without_run?
+          )
+
+        json(conn, %{work_order_id: work_order.id})
+
       {:error, reason, %{text: message}} ->
         status =
           if reason == :too_many_requests,
@@ -85,19 +95,7 @@ defmodule LightningWeb.WebhooksController do
 
         conn
         |> put_status(status)
-        |> json(%{"error" => message})
-
-      nil ->
-        conn
-        |> put_status(:not_found)
-        |> json(%{"error" => "Webhook not found"})
-
-      _disabled ->
-        put_status(conn, :forbidden)
-        |> json(%{
-          message:
-            "Unable to process request, trigger is disabled. Enable it on OpenFn to allow requests to this endpoint."
-        })
+        |> json(%{error: message})
     end
   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"},