diff --git a/CHANGELOG.txt b/CHANGELOG.txt index f3b4f8cf..52873af1 100644 --- a/CHANGELOG.txt +++ b/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 diff --git a/config/dev.exs b/config/dev.exs index 0a5748d1..db9d17f1 100644 --- a/config/dev.exs +++ b/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, diff --git a/lib/towerops/workers/check_executor_worker.ex b/lib/towerops/workers/check_executor_worker.ex index 4ec4e12d..53dce072 100644 --- a/lib/towerops/workers/check_executor_worker.ex +++ b/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 diff --git a/priv/static/changelog.txt b/priv/static/changelog.txt index 7ca73628..f123e2a6 100644 --- a/priv/static/changelog.txt +++ b/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 diff --git a/test/towerops/workers/check_executor_worker_test.exs b/test/towerops/workers/check_executor_worker_test.exs index 06e6ad5a..9ecc38d6 100644 --- a/test/towerops/workers/check_executor_worker_test.exs +++ b/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