add identity janitor

Author: lun-4

config/config.exs

@@ -14,7 +14,26 @@ config :vrhose,
 config :vrhose,
   ecto_repos: [VRHose.Repo]
 
-config :vrhose, VRHose.Repo, database: "vrhose_#{Mix.env()}.db"
+repos = [
+  VRHose.Repo,
+  VRHose.Repo.Replica1,
+  VRHose.Repo.Replica2,
+  VRHose.Repo.Replica3,
+  VRHose.Repo.Replica4,
+  VRHose.Repo.JanitorReplica
+]
+
+for repo <- repos do
+  config :vrhose, repo,
+    cache_size: -8_000,
+    pool_size: 1,
+    auto_vacuum: :incremental,
+    telemetry_prefix: [:vrhose, :repo],
+    telemetry_event: [VRHose.Repo.Instrumenter],
+    queue_target: 500,
+    queue_interval: 2000,
+    database: "vrhose_#{Mix.env()}.db"
+end
 
 # Configures the endpoint
 config :vrhose, VRHoseWeb.Endpoint,

lib/vrhose/application.ex

@@ -21,23 +21,25 @@ defmodule VRHose.Application do
 
     children =
       [
-        VRHose.Repo,
-        VRHoseWeb.Telemetry,
-        {VRHose.QuickLeader, name: VRHose.QuickLeader},
-        {Finch,
-         name: VRHose.Finch,
-         pools: %{
-           :default => [size: 50, count: 50]
-         }},
-        {DNSCluster, query: Application.get_env(:vrhose, :dns_cluster_query) || :ignore},
-        {Phoenix.PubSub, name: VRHose.PubSub},
-        {
-          Registry,
-          # , partitions: System.schedulers_online()},
-          keys: :duplicate, name: Registry.Timeliners
-        },
-        {ExHashRing.Ring, name: VRHose.Hydrator.Ring}
+        VRHoseWeb.Telemetry
       ] ++
+        repos() ++
+        [
+          {VRHose.QuickLeader, name: VRHose.QuickLeader},
+          {Finch,
+           name: VRHose.Finch,
+           pools: %{
+             :default => [size: 50, count: 50]
+           }},
+          {DNSCluster, query: Application.get_env(:vrhose, :dns_cluster_query) || :ignore},
+          {Phoenix.PubSub, name: VRHose.PubSub},
+          {
+            Registry,
+            # , partitions: System.schedulers_online()},
+            keys: :duplicate, name: Registry.Timeliners
+          },
+          {ExHashRing.Ring, name: VRHose.Hydrator.Ring}
+        ] ++
         hydration_workers() ++
         [
           {VRHose.Ingestor, name: {:global, VRHose.Ingestor}},
@@ -53,16 +55,51 @@ defmodule VRHose.Application do
             id: "websocket"
           },
           VRHoseWeb.Endpoint
-        ] ++ timeliner_workers()
+        ] ++ timeliner_workers() ++ janitor_workers()
 
     start_telemetry()
+    IO.inspect(children, label: "application tree")
 
     # See https://hexdocs.pm/elixir/Supervisor.html
     # for other strategies and supported options
     opts = [strategy: :one_for_one, name: VRHose.Supervisor, max_restarts: 10]
     Supervisor.start_link(children, opts)
   end
 
+  def primaries() do
+    Application.fetch_env!(:vrhose, :ecto_repos)
+  end
+
+  defp repos() do
+    Application.fetch_env!(:vrhose, :ecto_repos)
+    |> Enum.map(fn primary ->
+      primary
+      |> to_string
+      |> then(fn
+        "Elixir.VRHose.Repo" <> _ ->
+          spec = primary.repo_spec()
+          [primary] ++ spec.read_replicas ++ spec.dedicated_replicas
+
+        _ ->
+          []
+      end)
+    end)
+    |> Enum.reduce(fn x, acc -> x ++ acc end)
+    |> Enum.map(fn repo ->
+      case Application.fetch_env(:vrhose, repo) do
+        :error ->
+          raise RuntimeError, "Repo #{repo} not configured"
+
+        {:ok, cfg} ->
+          if Access.get(cfg, :database) == nil do
+            raise RuntimeError, "Repo #{repo} not configured. missing database"
+          end
+
+          repo
+      end
+    end)
+  end
+
   def hydration_workers() do
     1..20
     |> Enum.map(fn i ->
@@ -129,6 +166,19 @@ defmodule VRHose.Application do
     Logger.info("telemetry started!")
   end
 
+  defp janitor_specs do
+    [
+      [VRHose.Identity.Janitor, [every: 8 * 60, jitter: -60..60]]
+    ]
+  end
+
+  defp janitor_workers do
+    janitor_specs()
+    |> Enum.map(fn [module, opts] ->
+      VRHose.Tinycron.new(module, opts)
+    end)
+  end
+
   # Tell Phoenix to update the endpoint configuration
   # whenever the application is updated.
   @impl true

lib/vrhose/identity.ex

@@ -22,7 +22,7 @@ defmodule VRHose.Identity do
 
   def one(did) do
     query = from(s in __MODULE__, where: s.did == ^did, select: s)
-    Repo.one(query, log: false)
+    Repo.replica(did).one(query, log: false)
   end
 
   def fake(did) do
@@ -54,4 +54,47 @@ defmodule VRHose.Identity do
       log: false
     )
   end
+
+  defmodule Janitor do
+    require Logger
+
+    import Ecto.Query
+    alias VRHose.Identity
+    alias VRHose.Repo.JanitorReplica
+
+    def tick() do
+      Logger.info("cleaning identities...")
+
+      expiry_time =
+        NaiveDateTime.utc_now()
+        |> NaiveDateTime.add(-1, :day)
+        |> DateTime.from_naive!("Etc/UTC")
+        |> DateTime.to_unix()
+
+      deleted_count =
+        from(s in Identity,
+          where:
+            fragment("unixepoch(?)", s.inserted_at) <
+              ^expiry_time,
+          limit: 1000
+        )
+        |> JanitorReplica.all()
+        |> Enum.chunk_every(10)
+        |> Enum.map(fn chunk ->
+          chunk
+          |> Enum.map(fn identity ->
+            Repo.delete(identity)
+            1
+          end)
+          |> then(fn count ->
+            :timer.sleep(1500)
+            count
+          end)
+          |> Enum.sum()
+        end)
+        |> Enum.sum()
+
+      Logger.info("deleted #{deleted_count} identities")
+    end
+  end
 end

lib/vrhose/repo.ex

@@ -1,9 +1,21 @@
 defmodule VRHose.Repo do
-  use Ecto.Repo,
-    otp_app: :vrhose,
-    adapter: Ecto.Adapters.SQLite3,
-    pool_size: 1,
-    loggers: [VRHose.Repo.Instrumenter, Ecto.LogEntry]
+  use VRHose.Repo.Base,
+    primary: VRHose.Repo,
+    read_replicas: [
+      VRHose.Repo.Replica1,
+      VRHose.Repo.Replica2,
+      VRHose.Repo.Replica3,
+      VRHose.Repo.Replica4
+    ],
+    dedicated_replicas: [
+      VRHose.Repo.JanitorReplica
+    ]
+
+  # use Ecto.Repo,
+  #   otp_app: :vrhose,
+  #   adapter: Ecto.Adapters.SQLite3,
+  #   pool_size: 1,
+  #   loggers: [VRHose.Repo.Instrumenter, Ecto.LogEntry]
 
   defmodule Instrumenter do
     use Prometheus.EctoInstrumenter

lib/vrhose/repo_base.ex

@@ -0,0 +1,51 @@
+defmodule VRHose.Repo.Base do
+  defmacro __using__(opts) do
+    quote bind_quoted: [opts: opts] do
+      use Ecto.Repo,
+        otp_app: :vrhose,
+        adapter: Ecto.Adapters.SQLite3,
+
+        # sqlite does not do multi-writer. pool_size is effectively one,
+        # if it's larger than one, then Database Busy errors haunt you
+        # the trick to make concurrency happen is to create "read replicas"
+        # that are effectively a pool of readers. this works because we're in WAL mode
+        pool_size: 1,
+        loggers: [VRHose.Repo.Instrumenter, Ecto.LogEntry]
+
+      @read_replicas opts[:read_replicas]
+      @dedicated_replicas opts[:dedicated_replicas]
+
+      def repo_spec do
+        %{read_replicas: @read_replicas, dedicated_replicas: @dedicated_replicas}
+      end
+
+      def replica() do
+        Enum.random(@read_replicas)
+      end
+
+      def replica(identifier)
+          when is_number(identifier) or is_bitstring(identifier) or is_atom(identifier) do
+        @read_replicas |> Enum.at(rem(identifier |> :erlang.phash2(), length(@read_replicas)))
+      end
+
+      for repo <- @read_replicas ++ @dedicated_replicas do
+        default_dynamic_repo =
+          if Mix.env() == :test do
+            opts[:primary]
+          else
+            repo
+          end
+
+        defmodule repo do
+          use Ecto.Repo,
+            otp_app: :vrhose,
+            adapter: Ecto.Adapters.SQLite3,
+            pool_size: 1,
+            loggers: [VRHose.Repo.Instrumenter, Ecto.LogEntry],
+            read_only: true,
+            default_dynamic_repo: default_dynamic_repo
+        end
+      end
+    end
+  end
+end

lib/vrhose/tinycron.ex

@@ -0,0 +1,55 @@
+defmodule VRHose.Tinycron do
+  use GenServer
+  require Logger
+
+  def new(module, opts \\ []) do
+    %{
+      id: module,
+      start: {__MODULE__, :start_link, [module, opts |> Keyword.put(:name, module)]}
+    }
+  end
+
+  def start_link(module, opts \\ []) do
+    GenServer.start_link(__MODULE__, [module, opts], opts)
+  end
+
+  def noop(mod, value) do
+    GenServer.cast(mod, {:noop, value})
+  end
+
+  @impl true
+  def init([module, opts]) do
+    state = %{module: module, opts: opts}
+    schedule_work(state)
+    {:ok, state}
+  end
+
+  @impl true
+  def handle_info(:work, %{module: module} = state) do
+    unless state |> Map.get(:noop, false) do
+      Logger.debug("running #{inspect(state.module)}")
+      module.tick()
+    end
+
+    schedule_work(state)
+    {:noreply, state}
+  end
+
+  @impl true
+  def handle_cast({:noop, value}, state) do
+    {:noreply, state |> Map.put(:noop, value)}
+  end
+
+  defp schedule_work(state) do
+    every_seconds = state.opts |> Keyword.get(:every) || 10
+    jitter_seconds_range = state.opts |> Keyword.get(:jitter) || -2..2
+    first..last//step = jitter_seconds_range
+    # turn it into milliseconds for greater jitter possibilities
+    jitter_milliseconds_range = (first * 1000)..(last * 1000)//step
+
+    # prevent jitter from creating negative next_tick_time by doing max(0, next_tick_time)
+    next_tick_time = max(0, every_seconds * 1000 + Enum.random(jitter_milliseconds_range))
+    Logger.debug("scheduling #{inspect(state.module)} in #{next_tick_time}ms")
+    Process.send_after(self(), :work, next_tick_time)
+  end
+end