Upgrade Oban & configure Stager plugin (#1822)

config/runtime.exs

@@ -283,65 +283,81 @@ config :plausible, PlausibleWeb.Firewall,
     |> String.split(",")
     |> Enum.map(&String.trim/1)
 
-if config_env() == :prod && !disable_cron do
-  base_cron = [
-    # Daily at midnight
-    {"0 0 * * *", Plausible.Workers.RotateSalts},
-    #  hourly
-    {"0 * * * *", Plausible.Workers.ScheduleEmailReports},
-    # hourly
-    {"0 * * * *", Plausible.Workers.SendSiteSetupEmails},
-    # Daily at midday
-    {"0 12 * * *", Plausible.Workers.SendCheckStatsEmails},
-    # Every 15 minutes
-    {"*/15 * * * *", Plausible.Workers.SpikeNotifier},
-    # Every day at midnight
-    {"0 0 * * *", Plausible.Workers.CleanEmailVerificationCodes},
-    # Every day at 1am
-    {"0 1 * * *", Plausible.Workers.CleanInvitations}
-  ]
+base_cron = [
+  # Daily at midnight
+  {"0 0 * * *", Plausible.Workers.RotateSalts},
+  #  hourly
+  {"0 * * * *", Plausible.Workers.ScheduleEmailReports},
+  # hourly
+  {"0 * * * *", Plausible.Workers.SendSiteSetupEmails},
+  # Daily at midday
+  {"0 12 * * *", Plausible.Workers.SendCheckStatsEmails},
+  # Every 15 minutes
+  {"*/15 * * * *", Plausible.Workers.SpikeNotifier},
+  # Every day at midnight
+  {"0 0 * * *", Plausible.Workers.CleanEmailVerificationCodes},
+  # Every day at 1am
+  {"0 1 * * *", Plausible.Workers.CleanInvitations}
+]
 
-  extra_cron = [
-    # Daily at midday
-    {"0 12 * * *", Plausible.Workers.SendTrialNotifications},
-    # Daily at 14
-    {"0 14 * * *", Plausible.Workers.CheckUsage},
-    # Daily at 15
-    {"0 15 * * *", Plausible.Workers.NotifyAnnualRenewal},
-    # Every midnight
-    {"0 0 * * *", Plausible.Workers.LockSites}
-  ]
+extra_cron = [
+  # Daily at midday
+  {"0 12 * * *", Plausible.Workers.SendTrialNotifications},
+  # Daily at 14
+  {"0 14 * * *", Plausible.Workers.CheckUsage},
+  # Daily at 15
+  {"0 15 * * *", Plausible.Workers.NotifyAnnualRenewal},
+  # Every midnight
+  {"0 0 * * *", Plausible.Workers.LockSites}
+]
 
-  base_queues = [
-    rotate_salts: 1,
-    schedule_email_reports: 1,
-    send_email_reports: 1,
-    spike_notifications: 1,
-    check_stats_emails: 1,
-    site_setup_emails: 1,
-    clean_email_verification_codes: 1,
-    clean_invitations: 1,
-    google_analytics_imports: 1
-  ]
+queues = [
+  rotate_salts: 1,
+  schedule_email_reports: 1,
+  send_email_reports: 1,
+  spike_notifications: 1,
+  check_stats_emails: 1,
+  site_setup_emails: 1,
+  clean_email_verification_codes: 1,
+  clean_invitations: 1,
+  google_analytics_imports: 1,
+  test_job_retry: 1,
+  trial_notification_emails: 1,
+  check_usage: 1,
+  notify_annual_renewal: 1,
+  lock_sites: 1
+]
 
-  extra_queues = [
-    trial_notification_emails: 1,
-    check_usage: 1,
-    notify_annual_renewal: 1,
-    lock_sites: 1
-  ]
+cond do
+  config_env() == :prod && !disable_cron ->
+    config :plausible, Oban,
+      repo: Plausible.Repo,
+      plugins: [
+        # Keep 30 days history
+        {Oban.Plugins.Pruner, max_age: :timer.hours(24 * 30)},
+        {Oban.Plugins.Cron,
+         crontab: if(is_selfhost, do: base_cron, else: base_cron ++ extra_cron)},
+        # Rescue orphaned jobs after 2 hours
+        {Oban.Plugins.Lifeline, rescue_after: :timer.minutes(120)},
+        {Oban.Plugins.Stager, interval: :timer.seconds(5)}
+      ],
+      queues: queues
 
-  # Keep 30 days history
-  config :plausible, Oban,
-    repo: Plausible.Repo,
-    plugins: [{Oban.Plugins.Pruner, max_age: 2_592_000}],
-    queues: if(is_selfhost, do: base_queues, else: base_queues ++ extra_queues),
-    crontab: if(is_selfhost, do: base_cron, else: base_cron ++ extra_cron)
-else
-  config :plausible, Oban,
-    repo: Plausible.Repo,
-    queues: [google_analytics_imports: 1],
-    plugins: false
+  config_env() == :test ->
+    config :plausible, Oban,
+      repo: Plausible.Repo,
+      queues: queues,
+      plugins: false
+
+  true ->
+    config :plausible, Oban,
+      repo: Plausible.Repo,
+      queues: queues,
+      plugins: [
+        # Keep 1 day history
+        {Oban.Plugins.Pruner, max_age: :timer.hours(24)},
+        {Oban.Plugins.Stager, interval: :timer.seconds(5)}
+      ]
 end
 
 config :plausible, :hcaptcha,

lib/telemetry.ex

@@ -5,7 +5,7 @@ defmodule ErrorReporter do
       |> Map.take([:id, :args, :meta, :queue, :worker])
       |> Map.merge(measure)
 
-    maybe_log_import_error(job)
+    on_job_exception(job)
 
     Sentry.capture_exception(meta.error, stacktrace: meta.stacktrace, extra: extra)
   end
@@ -14,14 +14,28 @@ defmodule ErrorReporter do
     Sentry.capture_exception(meta.error, stacktrace: meta.stacktrace, extra: meta)
   end
 
-  defp maybe_log_import_error(%Oban.Job{
+  defp on_job_exception(%Oban.Job{
          queue: "google_analytics_imports",
-         args: %{"site_id" => site_id}
-       }) do
+         args: %{"site_id" => site_id},
+         state: "executing",
+         attempt: attempt,
+         max_attempts: max_attempts
+       })
+       when attempt >= max_attempts do
     site = Plausible.Repo.get(Plausible.Site, site_id)
 
     if site do
       Plausible.Workers.ImportGoogleAnalytics.import_failed(site)
     end
   end
+
+  defp on_job_exception(%Oban.Job{
+         queue: "google_analytics_imports",
+         args: %{"site_id" => site_id},
+         state: "executing"
+       }) do
+    Plausible.ClickhouseRepo.clear_imported_stats_for(site_id)
+  end
+
+  defp on_job_exception(_job), do: :ignore
 end

lib/workers/import_google_analytics.ex

@@ -3,7 +3,7 @@ defmodule Plausible.Workers.ImportGoogleAnalytics do
 
   use Oban.Worker,
     queue: :google_analytics_imports,
-    max_attempts: 1,
+    max_attempts: 3,
     unique: [fields: [:args], period: 60]
 
   @impl Oban.Worker
@@ -45,6 +45,12 @@ defmodule Plausible.Workers.ImportGoogleAnalytics do
     end
   end
 
+  @impl Oban.Worker
+  def backoff(_job) do
+    # 5 minutes
+    300
+  end
+
   def import_failed(site) do
     site = Repo.preload(site, memberships: :user)
 

mix.exs

@@ -52,7 +52,9 @@ defmodule Plausible.MixProject do
     [
       {:bcrypt_elixir, "~> 2.0"},
       {:combination, "~> 0.0.3"},
-      {:cors_plug, "~> 2.0"},
+      {:cors_plug, "~> 3.0"},
+      {:plug, "~> 1.13", override: true},
+      {:connection, "~> 1.1", override: true},
       {:ecto_sql, "~> 3.0"},
       {:elixir_uuid, "~> 1.2", only: :test},
       {:jason, "~> 1.3"},
@@ -62,7 +64,6 @@ defmodule Plausible.MixProject do
       {:phoenix_live_reload, "~> 1.2", only: :dev},
       {:phoenix_pubsub, "~> 2.0"},
       {:plug_cowboy, "~> 2.3"},
-      {:postgrex, ">= 0.0.0"},
       {:ref_inspector, "~> 1.3"},
       {:timex, "~> 3.6"},
       {:ua_inspector, "~> 3.0"},
@@ -82,7 +83,7 @@ defmodule Plausible.MixProject do
       {:oauther, "~> 1.3"},
       {:nanoid, "~> 2.0.2"},
       {:siphash, "~> 3.2"},
-      {:oban, "~> 2.0"},
+      {:oban, "~> 2.11"},
       {:geolix, "~> 1.0"},
       {:clickhouse_ecto, git: "https://github.com/plausible/clickhouse_ecto.git"},
       {:location, git: "https://github.com/plausible/location.git"},

mix.lock

@@ -14,8 +14,8 @@
   "combination": {:hex, :combination, "0.0.3", "746aedca63d833293ec6e835aa1f34974868829b1486b1e1cb0685f0b2ae1f41", [:mix], [], "hexpm", "72b099f463df42ef7dc6371d250c7070b57b6c5902853f69deb894f79eda18ca"},
   "combine": {:hex, :combine, "0.10.0", "eff8224eeb56498a2af13011d142c5e7997a80c8f5b97c499f84c841032e429f", [:mix], [], "hexpm", "1b1dbc1790073076580d0d1d64e42eae2366583e7aecd455d1215b0d16f2451b"},
   "comeonin": {:hex, :comeonin, "5.3.3", "2c564dac95a35650e9b6acfe6d2952083d8a08e4a89b93a481acb552b325892e", [:mix], [], "hexpm", "3e38c9c2cb080828116597ca8807bb482618a315bfafd98c90bc22a821cc84df"},
-  "connection": {:hex, :connection, "1.0.4", "a1cae72211f0eef17705aaededacac3eb30e6625b04a6117c1b2db6ace7d5976", [:mix], [], "hexpm", "4a0850c9be22a43af9920a71ab17c051f5f7d45c209e40269a1938832510e4d9"},
-  "cors_plug": {:hex, :cors_plug, "2.0.3", "316f806d10316e6d10f09473f19052d20ba0a0ce2a1d910ddf57d663dac402ae", [:mix], [{:plug, "~> 1.8", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "ee4ae1418e6ce117fc42c2ba3e6cbdca4e95ecd2fe59a05ec6884ca16d469aea"},
+  "connection": {:hex, :connection, "1.1.0", "ff2a49c4b75b6fb3e674bfc5536451607270aac754ffd1bdfe175abe4a6d7a68", [:mix], [], "hexpm", "722c1eb0a418fbe91ba7bd59a47e28008a189d47e37e0e7bb85585a016b2869c"},
+  "cors_plug": {:hex, :cors_plug, "3.0.3", "7c3ac52b39624bc616db2e937c282f3f623f25f8d550068b6710e58d04a0e330", [:mix], [{:plug, "~> 1.13", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "3f2d759e8c272ed3835fab2ef11b46bddab8c1ab9528167bd463b6452edf830d"},
   "cowboy": {:hex, :cowboy, "2.9.0", "865dd8b6607e14cf03282e10e934023a1bd8be6f6bacf921a7e2a96d800cd452", [:make, :rebar3], [{:cowlib, "2.11.0", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, "1.8.0", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm", "2c729f934b4e1aa149aff882f57c6372c15399a20d54f65c8d67bef583021bde"},
   "cowboy_telemetry": {:hex, :cowboy_telemetry, "0.3.1", "ebd1a1d7aff97f27c66654e78ece187abdc646992714164380d8a041eda16754", [:rebar3], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "3a6efd3366130eab84ca372cbd4a7d3c3a97bdfcfb4911233b035d117063f0af"},
   "cowlib": {:hex, :cowlib, "2.11.0", "0b9ff9c346629256c42ebe1eeb769a83c6cb771a6ee5960bd110ab0b9b872063", [:make, :rebar3], [], "hexpm", "2b3e9da0b21c4565751a6d4901c20d1b4cc25cbb7fd50d91d2ab6dd287bc86a9"},
@@ -64,7 +64,7 @@
   "mmdb2_decoder": {:hex, :mmdb2_decoder, "3.0.0", "54828676a36e75e9a25bc9a0bb0598d4c7fcc767bf0b40674850b22e05b7b6cc", [:mix], [], "hexpm", "359dc9242915538d1dceb9f6d96c72201dca76ce62e49d22e2ed1e86f20bea8e"},
   "nanoid": {:hex, :nanoid, "2.0.5", "1d2948d8967ef2d948a58c3fef02385040bd9823fc6394bd604b8d98e5516b22", [:mix], [], "hexpm", "956e8876321104da72aa48770539ff26b36b744cd26753ec8e7a8a37e53d5f58"},
   "oauther": {:hex, :oauther, "1.3.0", "82b399607f0ca9d01c640438b34d74ebd9e4acd716508f868e864537ecdb1f76", [:mix], [], "hexpm", "78eb888ea875c72ca27b0864a6f550bc6ee84f2eeca37b093d3d833fbcaec04e"},
-  "oban": {:hex, :oban, "2.10.1", "202a90f2aed0130b7d750bdbfea8090c8321bce255bade10fd3699733565add0", [:mix], [{:ecto_sql, "~> 3.6", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.14", [hex: :postgrex, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "161cdd01194147cd6a3efdb1d6c3d9689309991412f799c1e242c18912e307c3"},
+  "oban": {:hex, :oban, "2.11.3", "f431f2f0c251b8490a7fa00d2cce7197a1cf4d3f04a3305e80411f083392998f", [:mix], [{:ecto_sql, "~> 3.6", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.16", [hex: :postgrex, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "26529da52bfca27740c984bfc70e7f954d6411ceeae0c7c94d2c8aa7c00b513d"},
   "parallel_stream": {:hex, :parallel_stream, "1.0.6", "b967be2b23f0f6787fab7ed681b4c45a215a81481fb62b01a5b750fa8f30f76c", [:mix], [], "hexpm", "639b2e8749e11b87b9eb42f2ad325d161c170b39b288ac8d04c4f31f8f0823eb"},
   "parse_trans": {:hex, :parse_trans, "3.3.1", "16328ab840cc09919bd10dab29e431da3af9e9e7e7e6f0089dd5a2d2820011d8", [:rebar3], [], "hexpm", "07cd9577885f56362d414e8c4c4e6bdf10d43a8767abb92d24cbe8b24c54888b"},
   "phoenix": {:hex, :phoenix, "1.6.6", "281c8ce8dccc9f60607346b72cdfc597c3dde134dd9df28dff08282f0b751754", [:mix], [{:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:phoenix_pubsub, "~> 2.0", [hex: :phoenix_pubsub, repo: "hexpm", optional: false]}, {:phoenix_view, "~> 1.0", [hex: :phoenix_view, repo: "hexpm", optional: false]}, {:plug, "~> 1.10", [hex: :plug, repo: "hexpm", optional: false]}, {:plug_cowboy, "~> 2.2", [hex: :plug_cowboy, repo: "hexpm", optional: true]}, {:plug_crypto, "~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "807bd646e64cd9dc83db016199715faba72758e6db1de0707eef0a2da4924364"},
@@ -75,12 +75,12 @@
   "phoenix_pubsub": {:hex, :phoenix_pubsub, "2.1.1", "ba04e489ef03763bf28a17eb2eaddc2c20c6d217e2150a61e3298b0f4c2012b5", [:mix], [], "hexpm", "81367c6d1eea5878ad726be80808eb5a787a23dee699f96e72b1109c57cdd8d9"},
   "phoenix_view": {:hex, :phoenix_view, "1.1.2", "1b82764a065fb41051637872c7bd07ed2fdb6f5c3bd89684d4dca6e10115c95a", [:mix], [{:phoenix_html, "~> 2.14.2 or ~> 3.0", [hex: :phoenix_html, repo: "hexpm", optional: true]}], "hexpm", "7ae90ad27b09091266f6adbb61e1d2516a7c3d7062c6789d46a7554ec40f3a56"},
   "php_serializer": {:hex, :php_serializer, "2.0.0", "b43f31aca22ed7321f32da2b94fe2ddf9b6739a965cb51541969119e572e821d", [:mix], [], "hexpm", "61e402e99d9062c0225a3f4fcf7e43b4cba1b8654944c0e7c139c3ca9de481da"},
-  "plug": {:hex, :plug, "1.11.1", "f2992bac66fdae679453c9e86134a4201f6f43a687d8ff1cd1b2862d53c80259", [:mix], [{:mime, "~> 1.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "23524e4fefbb587c11f0833b3910bfb414bf2e2534d61928e920f54e3a1b881f"},
+  "plug": {:hex, :plug, "1.13.4", "addb6e125347226e3b11489e23d22a60f7ab74786befb86c14f94fb5f23ca9a4", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "06114c1f2a334212fe3ae567dbb3b1d29fd492c1a09783d52f3d489c1a6f4cf2"},
   "plug_cowboy": {:hex, :plug_cowboy, "2.5.2", "62894ccd601cf9597e2c23911ff12798a8a18d237e9739f58a6b04e4988899fe", [:mix], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:cowboy_telemetry, "~> 0.3", [hex: :cowboy_telemetry, repo: "hexpm", optional: false]}, {:plug, "~> 1.7", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm", "ea6e87f774c8608d60c8d34022a7d073bd7680a0a013f049fc62bf35efea1044"},
   "plug_crypto": {:hex, :plug_crypto, "1.2.2", "05654514ac717ff3a1843204b424477d9e60c143406aa94daf2274fdd280794d", [:mix], [], "hexpm", "87631c7ad914a5a445f0a3809f99b079113ae4ed4b867348dd9eec288cecb6db"},
   "poison": {:hex, :poison, "3.1.0", "d9eb636610e096f86f25d9a46f35a9facac35609a7591b3be3326e99a0484665", [:mix], [], "hexpm", "fec8660eb7733ee4117b85f55799fd3833eb769a6df71ccf8903e8dc5447cfce"},
   "poolboy": {:hex, :poolboy, "1.5.2", "392b007a1693a64540cead79830443abf5762f5d30cf50bc95cb2c1aaafa006b", [:rebar3], [], "hexpm", "dad79704ce5440f3d5a3681c8590b9dc25d1a561e8f5a9c995281012860901e3"},
-  "postgrex": {:hex, :postgrex, "0.15.13", "7794e697481799aee8982688c261901de493eb64451feee6ea58207d7266d54a", [:mix], [{:connection, "~> 1.0", [hex: :connection, repo: "hexpm", optional: false]}, {:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm", "3ffb76e1a97cfefe5c6a95632a27ffb67f28871c9741fb585f9d1c3cd2af70f1"},
+  "postgrex": {:hex, :postgrex, "0.16.2", "0f83198d0e73a36e8d716b90f45f3bde75b5eebf4ade4f43fa1f88c90a812f74", [:mix], [{:connection, "~> 1.1", [hex: :connection, repo: "hexpm", optional: false]}, {:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm", "a9ea589754d9d4d076121090662b7afe155b374897a6550eb288f11d755acfa0"},
   "public_suffix": {:git, "https://github.com/axelson/publicsuffix-elixir", "89372422ab8b433de508519ef474e39699fd11ca", []},
   "ranch": {:hex, :ranch, "1.8.0", "8c7a100a139fd57f17327b6413e4167ac559fbc04ca7448e9be9057311597a1d", [:make, :rebar3], [], "hexpm", "49fbcfd3682fab1f5d109351b61257676da1a2fdbe295904176d5e521a2ddfe5"},
   "ref_inspector": {:hex, :ref_inspector, "1.3.1", "bb0489a4c4299dcd633f2b7a60c41a01f5590789d0b28225a60be484e1fbe777", [:mix], [{:hackney, "~> 1.0", [hex: :hackney, repo: "hexpm", optional: false]}, {:yamerl, "~> 0.7", [hex: :yamerl, repo: "hexpm", optional: false]}], "hexpm", "3172eb1b08e5c69966f796e3fe0e691257546fa143a5eb0ecc18a6e39b233854"},

priv/repo/migrations/20220408071645_create_oban_peers.exs

@@ -0,0 +1,6 @@
+defmodule Plausible.Repo.Migrations.CreateObanPeers do
+  use Ecto.Migration
+
+  def up, do: Oban.Migrations.up(version: 11)
+  def down, do: Oban.Migrations.down(version: 11)
+end

priv/repo/migrations/20220408080058_swap_primary_oban_indexes.exs

@@ -0,0 +1,21 @@
+defmodule Plausible.Repo.Migrations.SwapPrimaryObanIndexes do
+  use Ecto.Migration
+
+  @disable_ddl_transaction true
+  @disable_migration_lock true
+
+  def change do
+    create_if_not_exists index(
+                           :oban_jobs,
+                           [:state, :queue, :priority, :scheduled_at, :id],
+                           concurrently: true,
+                           prefix: "public"
+                         )
+
+    drop_if_exists index(
+                     :oban_jobs,
+                     [:queue, :state, :priority, :scheduled_at, :id],
+                     prefix: "public"
+                   )
+  end
+end