fix(monitoring): stop CheckExecutorWorker duplicate job buildup with unique constraint

CheckExecutorWorker had no Oban `unique` constraint, so every insert path
created a fresh job row: self-scheduling, JobHealthCheckWorker, the check
form/API/GraphQL, and — the main amplifier — Monitoring.schedule_check/1
called unconditionally by snmp.ex discovery after create_check/1, which is
an upsert that returns {:ok, check} even when the check already exists.
Every SNMP discovery pass re-enqueued a job for every existing check, so
the check_executors queue grew to ~70k jobs for ~5.7k checks and stopped
draining.

Add `unique: [keys: [:check_id], states: [:available, :scheduled,
:retryable], period: :infinity]`. `:executing` is excluded so perform/1's
self-reschedule does not conflict with its own running job.

Also add the `checks` and `check_executors` queues to config/dev.exs —
they were missing from dev entirely (prod's runtime.exs already had them).

CHANGELOG.txt

@@ -1,4 +1,23 @@
 2026-05-14
+fix(monitoring): add Oban unique constraint to CheckExecutorWorker to stop duplicate job buildup
+  CheckExecutorWorker had no `unique` constraint. Every insert path —
+  self-scheduling (schedule_next_check/1), Monitoring.schedule_check/1 called
+  unconditionally by snmp.ex discovery (create_check/1 is an upsert that returns
+  {:ok, check} even for existing checks), JobHealthCheckWorker, the check form,
+  API v1 and GraphQL — created a fresh job row. Every SNMP discovery pass
+  re-enqueued a job for every existing check, so the check_executors queue grew
+  to ~70k jobs for ~5.7k checks (~12-45 dupes/check) and could not drain.
+  - lib/towerops/workers/check_executor_worker.ex: added
+    `unique: [keys: [:check_id], states: [:available, :scheduled, :retryable],
+    period: :infinity]`. `:executing` is excluded so perform/1's self-reschedule
+    does not conflict with its own running job.
+  - config/dev.exs: added the `checks` and `check_executors` Oban queues, which
+    were missing from dev (prod's runtime.exs already had them) — dev never
+    processed those queues at all.
+  - test/towerops/workers/check_executor_worker_test.exs: added "job uniqueness"
+    tests covering dedup on insert, dedup via schedule_check/1, and that an
+    existing queued job does not block perform/1 from rescheduling.
+
 fix(sync): classify unexpected 5xx statuses as transient so vendor syncs back off and retry
   Vendor API clients (Preseem, etc.) return errors shaped as
   `{:unexpected_status, status, body}`, but `SyncErrors.transient?/1` only

config/dev.exs

@@ -62,6 +62,9 @@ config :towerops, Oban,
     pollers: 50,
     # Device monitoring jobs - health checks
     monitors: 50,
+    # Service checks - HTTP/TCP/DNS
+    checks: 50,
+    check_executors: 50,
     # Alert notifications (PagerDuty, email, etc.)
     notifications: 10,
     maintenance: 5,

lib/towerops/workers/check_executor_worker.ex

@@ -19,9 +19,20 @@ defmodule Towerops.Workers.CheckExecutorWorker do
   polling offsets to prevent thundering herd.
   """
 
+  # `unique` keeps at most one pending job per check_id. Every insert path
+  # (self-scheduling, discovery re-runs, JobHealthCheckWorker, the check form,
+  # API and GraphQL) goes through this worker, so without it duplicate jobs
+  # accumulate without bound. `:executing` is deliberately excluded: perform/1
+  # self-schedules the next run while the current job is still executing, and
+  # including it would make that reschedule conflict with itself.
   use Oban.Worker,
     queue: :check_executors,
-    max_attempts: 3
+    max_attempts: 3,
+    unique: [
+      keys: [:check_id],
+      states: [:available, :scheduled, :retryable],
+      period: :infinity
+    ]
 
   alias Towerops.Agents
   alias Towerops.Monitoring

priv/static/changelog.txt

@@ -1,3 +1,7 @@
+2026-05-14 — Monitoring check reliability fix
+* Fixed an issue where monitoring checks could build up a large backlog of duplicate scheduled runs, delaying check execution
+* Checks now stay on schedule instead of falling behind
+
 2026-05-10 — AI now spots patterns rule-based monitoring misses
 * The AI now reviews your network as a whole — looking at QoE, signal, capacity, agents, reconciliation findings, and the insights already surfaced — and adds its own observations when it spots something the rule-based monitors don't already cover
 * AI-generated observations appear alongside existing insights, tagged with an "AI" source badge so you can tell where each finding came from

test/towerops/workers/check_executor_worker_test.exs

@@ -577,4 +577,100 @@ defmodule Towerops.Workers.CheckExecutorWorkerTest do
       assert hd(jobs).args["check_id"] == check.id
     end
   end
+
+  describe "perform/1 - job uniqueness" do
+    test "inserting two jobs for the same check_id keeps only one queued job", %{device: device} do
+      {:ok, check} =
+        Monitoring.create_check(%{
+          organization_id: device.organization_id,
+          device_id: device.id,
+          name: "Uniqueness Check",
+          check_type: "http",
+          source_type: "manual",
+          interval_seconds: 60,
+          enabled: true,
+          config: %{"url" => "https://example.com"}
+        })
+
+      Repo.delete_all(Oban.Job)
+
+      assert {:ok, job1} = %{check_id: check.id} |> CheckExecutorWorker.new() |> Oban.insert()
+      assert {:ok, job2} = %{check_id: check.id} |> CheckExecutorWorker.new() |> Oban.insert()
+
+      # Second insert is deduplicated against the first, not a new row
+      assert job2.id == job1.id
+      assert job2.conflict?
+
+      jobs =
+        Repo.all(
+          from j in Oban.Job,
+            where: j.worker == "Towerops.Workers.CheckExecutorWorker",
+            where: fragment("?->>'check_id'", j.args) == ^check.id
+        )
+
+      assert length(jobs) == 1
+    end
+
+    test "Monitoring.schedule_check/1 called repeatedly queues only one job", %{device: device} do
+      {:ok, check} =
+        Monitoring.create_check(%{
+          organization_id: device.organization_id,
+          device_id: device.id,
+          name: "Schedule Check Dedup",
+          check_type: "http",
+          source_type: "manual",
+          interval_seconds: 60,
+          enabled: true,
+          config: %{"url" => "https://example.com"}
+        })
+
+      Repo.delete_all(Oban.Job)
+
+      assert {:ok, _} = Monitoring.schedule_check(check)
+      assert {:ok, _} = Monitoring.schedule_check(check)
+      assert {:ok, _} = Monitoring.schedule_check(check)
+
+      jobs =
+        Repo.all(
+          from j in Oban.Job,
+            where: j.worker == "Towerops.Workers.CheckExecutorWorker",
+            where: fragment("?->>'check_id'", j.args) == ^check.id
+        )
+
+      assert length(jobs) == 1
+    end
+
+    test "an existing queued job does not block perform/1 from rescheduling", %{device: device} do
+      {:ok, check} =
+        Monitoring.create_check(%{
+          organization_id: device.organization_id,
+          device_id: device.id,
+          name: "Reschedule Through Existing",
+          check_type: "http",
+          source_type: "manual",
+          interval_seconds: 60,
+          enabled: true,
+          config: %{"url" => "https://example.com"}
+        })
+
+      Repo.delete_all(Oban.Job)
+
+      # A job is already queued for this check (e.g. from discovery)
+      assert {:ok, _} = Monitoring.schedule_check(check)
+
+      # Running the check still succeeds and self-scheduling still leaves
+      # exactly one pending job (the executing job is excluded from the
+      # uniqueness states, so the reschedule is not blocked).
+      assert :ok = perform_job(CheckExecutorWorker, %{check_id: check.id})
+
+      jobs =
+        Repo.all(
+          from j in Oban.Job,
+            where: j.worker == "Towerops.Workers.CheckExecutorWorker",
+            where: fragment("?->>'check_id'", j.args) == ^check.id
+        )
+
+      assert length(jobs) == 1
+    end
+  end
 end