From 52cde2cbc5342f60c27936cb617cf9e900c0bf5a Mon Sep 17 00:00:00 2001
From: Graham McIntire <graham@mcintire.me>
Date: Tue, 12 May 2026 12:12:11 -0500
Subject: [PATCH] =?UTF-8?q?fix:=20H21=20=E2=80=94=20stagger=20per-integrat?=
 =?UTF-8?q?ion=20sync=20jobs=20across=20maintenance=20queue?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Sonar, Splynx, VISP, NetBox, and Gaiia sync workers all followed the same
"cron runs, then Enum.map syncs every integration in the same job" shape.
A single slow tenant — anything from an upstream rate limit to a stuck
GraphQL call — would hold the maintenance queue slot for ten minutes
while every other integration waited in line.

Each worker now uses the dispatcher/per-integration pattern that
UispSyncWorker already had:

- the cron `perform(%Oban.Job{args: %{}})` fans out one job per eligible
  integration via `PollingOffset.calculate_offset/2`, staggered across a
  5- to 10-minute window with `unique: [period: window_seconds]`
- `perform(%Oban.Job{args: %{"integration_id" => id}})` does the actual
  sync, classifying failures via `Towerops.Workers.SyncErrors.transient?/1`
  so permanent errors (auth, 4xx) don't get retried forever

Worker-specific defaults preserved:
- gaiia / netbox: 30/600s window, longer interval
- sonar / splynx / visp: 10/300s window

Tests updated to match the two-stage dispatcher pattern — `perform_job/2`
with empty args now asserts the per-integration jobs got enqueued and
the actual sync is invoked via a separate `perform_job/2` call with the
integration_id arg.
---
 bugs.md                                       | 17 ----
 lib/towerops/workers/gaiia_sync_worker.ex     | 84 +++++++++++++------
 lib/towerops/workers/netbox_sync_worker.ex    | 84 +++++++++++++------
 lib/towerops/workers/sonar_sync_worker.ex     | 82 +++++++++++++-----
 lib/towerops/workers/splynx_sync_worker.ex    | 80 +++++++++++++-----
 lib/towerops/workers/visp_sync_worker.ex      | 80 +++++++++++++-----
 .../workers/gaiia_sync_worker_test.exs        | 39 +++++++--
 .../workers/sync_worker_perform_test.exs      | 44 +++++-----
 8 files changed, 349 insertions(+), 161 deletions(-)

diff --git a/bugs.md b/bugs.md
index 5eb67a56..4a689d2b 100644
--- a/bugs.md
+++ b/bugs.md
@@ -55,23 +55,6 @@
 
 ---
 
-### H21. Unbounded Sequential Processing in Sync Workers
-
-**Files:**
-- `gaiia_sync_worker.ex:20`
-- `sonar_sync_worker.ex:16`
-- `splynx_sync_worker.ex:16`
-- `visp_sync_worker.ex:16`
-- `netbox_sync_worker.ex:20`
-
-**Severity:** HIGH — Single job processes ALL integrations sequentially; 10+ minute blocks
-
-**Description:** `Enum.map(integrations, &sync_integration/1)` — sequentially syncs every enabled integration in a single Oban job. Blocks the queue slot for extended periods.
-
-**Fix:** Follow `UispSyncWorker` pattern — staggered per-integration jobs with unique constraints.
-
----
-
 ### H22. DataRetentionWorker Unbounded DELETEs
 
 **File:** `lib/towerops/workers/data_retention_worker.ex:61-173`
diff --git a/lib/towerops/workers/gaiia_sync_worker.ex b/lib/towerops/workers/gaiia_sync_worker.ex
index 02c6fcac..ce2b8acb 100644
--- a/lib/towerops/workers/gaiia_sync_worker.ex
+++ b/lib/towerops/workers/gaiia_sync_worker.ex
@@ -1,50 +1,82 @@
 defmodule Towerops.Workers.GaiiaSyncWorker do
   @moduledoc """
-  Oban cron worker that syncs Gaiia data for all enabled integrations.
+  Oban worker that syncs Gaiia data for enabled integrations.
 
-  Runs every 15 minutes. For each org with an enabled Gaiia integration,
-  checks if enough time has elapsed since the last sync (based on the
-  integration's sync_interval_minutes), then calls Gaiia.Sync.sync_organization/1.
+  The cron job (no args) runs every 15 minutes and dispatches individual
+  per-integration sync jobs staggered across the 600-second window using
+  `PollingOffset.calculate_offset/2`. This prevents a thundering herd
+  against the Gaiia GraphQL API and stops one slow tenant from blocking
+  the maintenance queue while every other integration waits in line.
+
+  Individual jobs receive `%{"integration_id" => id}` and perform the
+  actual sync via `Gaiia.Sync.sync_organization/1`.
   """
   use Oban.Worker, queue: :maintenance
 
   alias Towerops.Gaiia.Sync
   alias Towerops.Integrations
+  alias Towerops.Workers.PollingOffset
+  alias Towerops.Workers.SyncErrors
 
   require Logger
 
+  @provider "gaiia"
+  @default_interval_minutes 15
+  @window_seconds 600
+
+  # Cron dispatcher: enqueues staggered individual sync jobs.
   @impl Oban.Worker
-  def perform(%Oban.Job{}) do
-    integrations = Integrations.list_enabled_integrations("gaiia")
+  def perform(%Oban.Job{args: args}) when args == %{} do
+    integrations = Integrations.list_enabled_integrations(@provider)
 
-    results = Enum.map(integrations, &sync_integration/1)
+    eligible = Enum.filter(integrations, &due_for_sync?/1)
 
-    synced = Enum.count(results, &match?({:ok, _}, &1))
-    failed = Enum.count(results, &match?({:error, _}, &1))
-    skipped = Enum.count(results, &(&1 == :skipped))
+    now = DateTime.utc_now()
 
-    if synced > 0 or failed > 0 do
-      Logger.info("Gaiia sync batch: #{synced} synced, #{failed} failed, #{skipped} skipped")
+    enqueued =
+      Enum.count(eligible, fn integration ->
+        offset = PollingOffset.calculate_offset(integration.organization_id, @window_seconds)
+        scheduled_at = DateTime.add(now, offset, :second)
+
+        case %{"integration_id" => integration.id}
+             |> new(scheduled_at: scheduled_at, unique: [period: @window_seconds])
+             |> Oban.insert() do
+          {:ok, _job} -> true
+          {:error, _reason} -> false
+        end
+      end)
+
+    if enqueued > 0 do
+      Logger.info("Gaiia sync: dispatched #{enqueued} jobs across #{@window_seconds}s window")
     end
 
     :ok
   end
 
-  defp sync_integration(integration) do
-    if Integrations.due_for_sync?(integration, 15) do
-      case Sync.sync_organization(integration) do
-        {:ok, result} ->
-          Logger.info("Gaiia sync completed for org #{integration.organization_id}: #{inspect(result)}")
+  # Individual sync: loads integration by ID and syncs.
+  def perform(%Oban.Job{args: %{"integration_id" => id}}) do
+    case Integrations.get_integration_by_id(id) do
+      {:ok, integration} ->
+        sync_integration(integration)
 
-          {:ok, result}
-
-        {:error, reason} ->
-          Logger.error("Gaiia sync failed for org #{integration.organization_id}: #{inspect(reason)}")
-
-          {:error, reason}
-      end
-    else
-      :skipped
+      {:error, :not_found} ->
+        Logger.warning("Gaiia sync: integration #{id} not found, skipping")
+        {:error, :not_found}
     end
   end
+
+  defp sync_integration(integration) do
+    case Sync.sync_organization(integration) do
+      {:ok, result} ->
+        Logger.info("Gaiia sync completed for org #{integration.organization_id}: #{inspect(result)}")
+        :ok
+
+      {:error, reason} ->
+        Logger.error("Gaiia sync failed for org #{integration.organization_id}: #{inspect(reason)}")
+        if SyncErrors.transient?(reason), do: {:error, reason}, else: :ok
+    end
+  end
+
+  @doc false
+  def due_for_sync?(integration), do: Integrations.due_for_sync?(integration, @default_interval_minutes)
 end
diff --git a/lib/towerops/workers/netbox_sync_worker.ex b/lib/towerops/workers/netbox_sync_worker.ex
index df029ed2..4ebc88be 100644
--- a/lib/towerops/workers/netbox_sync_worker.ex
+++ b/lib/towerops/workers/netbox_sync_worker.ex
@@ -1,50 +1,82 @@
 defmodule Towerops.Workers.NetBoxSyncWorker do
   @moduledoc """
-  Oban cron worker that syncs NetBox data for all enabled integrations.
+  Oban worker that syncs NetBox data for enabled integrations.
 
-  Runs every 30 minutes. For each org with an enabled NetBox integration,
-  checks if enough time has elapsed since the last sync (based on the
-  integration's sync_interval_minutes), then calls NetBox.Sync.sync_organization/1.
+  The cron job (no args) runs every 30 minutes and dispatches individual
+  per-integration sync jobs staggered across the 600-second window using
+  `PollingOffset.calculate_offset/2`. This prevents a thundering herd
+  against NetBox APIs and stops one slow tenant from blocking the
+  maintenance queue while every other integration waits in line.
+
+  Individual jobs receive `%{"integration_id" => id}` and perform the
+  actual sync via `NetBox.Sync.sync_organization/1`.
   """
   use Oban.Worker, queue: :maintenance
 
   alias Towerops.Integrations
   alias Towerops.NetBox.Sync
+  alias Towerops.Workers.PollingOffset
+  alias Towerops.Workers.SyncErrors
 
   require Logger
 
+  @provider "netbox"
+  @default_interval_minutes 30
+  @window_seconds 600
+
+  # Cron dispatcher: enqueues staggered individual sync jobs.
   @impl Oban.Worker
-  def perform(%Oban.Job{}) do
-    integrations = Integrations.list_enabled_integrations("netbox")
+  def perform(%Oban.Job{args: args}) when args == %{} do
+    integrations = Integrations.list_enabled_integrations(@provider)
 
-    results = Enum.map(integrations, &sync_integration/1)
+    eligible = Enum.filter(integrations, &due_for_sync?/1)
 
-    synced = Enum.count(results, &match?({:ok, _}, &1))
-    failed = Enum.count(results, &match?({:error, _}, &1))
-    skipped = Enum.count(results, &(&1 == :skipped))
+    now = DateTime.utc_now()
 
-    if synced > 0 or failed > 0 do
-      Logger.info("NetBox sync batch: #{synced} synced, #{failed} failed, #{skipped} skipped")
+    enqueued =
+      Enum.count(eligible, fn integration ->
+        offset = PollingOffset.calculate_offset(integration.organization_id, @window_seconds)
+        scheduled_at = DateTime.add(now, offset, :second)
+
+        case %{"integration_id" => integration.id}
+             |> new(scheduled_at: scheduled_at, unique: [period: @window_seconds])
+             |> Oban.insert() do
+          {:ok, _job} -> true
+          {:error, _reason} -> false
+        end
+      end)
+
+    if enqueued > 0 do
+      Logger.info("NetBox sync: dispatched #{enqueued} jobs across #{@window_seconds}s window")
     end
 
     :ok
   end
 
-  defp sync_integration(integration) do
-    if Integrations.due_for_sync?(integration, 30) do
-      case Sync.sync_organization(integration) do
-        {:ok, result} ->
-          Logger.info("NetBox sync completed for org #{integration.organization_id}: #{inspect(result)}")
+  # Individual sync: loads integration by ID and syncs.
+  def perform(%Oban.Job{args: %{"integration_id" => id}}) do
+    case Integrations.get_integration_by_id(id) do
+      {:ok, integration} ->
+        sync_integration(integration)
 
-          {:ok, result}
-
-        {:error, reason} ->
-          Logger.error("NetBox sync failed for org #{integration.organization_id}: #{inspect(reason)}")
-
-          {:error, reason}
-      end
-    else
-      :skipped
+      {:error, :not_found} ->
+        Logger.warning("NetBox sync: integration #{id} not found, skipping")
+        {:error, :not_found}
     end
   end
+
+  defp sync_integration(integration) do
+    case Sync.sync_organization(integration) do
+      {:ok, result} ->
+        Logger.info("NetBox sync completed for org #{integration.organization_id}: #{inspect(result)}")
+        :ok
+
+      {:error, reason} ->
+        Logger.error("NetBox sync failed for org #{integration.organization_id}: #{inspect(reason)}")
+        if SyncErrors.transient?(reason), do: {:error, reason}, else: :ok
+    end
+  end
+
+  @doc false
+  def due_for_sync?(integration), do: Integrations.due_for_sync?(integration, @default_interval_minutes)
 end
diff --git a/lib/towerops/workers/sonar_sync_worker.ex b/lib/towerops/workers/sonar_sync_worker.ex
index 45f8d03b..4a2e2f49 100644
--- a/lib/towerops/workers/sonar_sync_worker.ex
+++ b/lib/towerops/workers/sonar_sync_worker.ex
@@ -1,44 +1,84 @@
 defmodule Towerops.Workers.SonarSyncWorker do
   @moduledoc """
-  Oban cron worker that syncs Sonar data for all enabled integrations.
+  Oban worker that syncs Sonar data for enabled integrations.
+
+  The cron job (no args) runs every 5 minutes and dispatches individual
+  per-integration sync jobs staggered across the 300-second window using
+  `PollingOffset.calculate_offset/2`. This prevents a thundering herd
+  against Sonar APIs and stops one slow tenant from blocking the
+  maintenance queue while every other integration waits in line.
+
+  Individual jobs receive `%{"integration_id" => id}` and perform the
+  actual sync via `Sonar.Sync.sync_organization/1`.
   """
   use Oban.Worker, queue: :maintenance
 
   alias Towerops.Integrations
   alias Towerops.Sonar.Sync
+  alias Towerops.Workers.PollingOffset
+  alias Towerops.Workers.SyncErrors
 
   require Logger
 
+  @provider "sonar"
+  @default_interval_minutes 10
+  @window_seconds 300
+
+  # Cron dispatcher: enqueues staggered individual sync jobs.
   @impl Oban.Worker
-  def perform(%Oban.Job{}) do
-    integrations = Integrations.list_enabled_integrations("sonar")
+  def perform(%Oban.Job{args: args}) when args == %{} do
+    integrations = Integrations.list_enabled_integrations(@provider)
 
-    results = Enum.map(integrations, &sync_integration/1)
+    eligible = Enum.filter(integrations, &due_for_sync?/1)
 
-    synced = Enum.count(results, &match?({:ok, _}, &1))
-    failed = Enum.count(results, &match?({:error, _}, &1))
-    skipped = Enum.count(results, &(&1 == :skipped))
+    now = DateTime.utc_now()
 
-    if synced > 0 or failed > 0 do
-      Logger.info("Sonar sync batch: #{synced} synced, #{failed} failed, #{skipped} skipped")
+    enqueued =
+      Enum.count(eligible, fn integration ->
+        offset = PollingOffset.calculate_offset(integration.organization_id, @window_seconds)
+        scheduled_at = DateTime.add(now, offset, :second)
+
+        case %{"integration_id" => integration.id}
+             |> new(scheduled_at: scheduled_at, unique: [period: @window_seconds])
+             |> Oban.insert() do
+          {:ok, _job} -> true
+          {:error, _reason} -> false
+        end
+      end)
+
+    if enqueued > 0 do
+      Logger.info("Sonar sync: dispatched #{enqueued} jobs across #{@window_seconds}s window")
     end
 
     :ok
   end
 
-  defp sync_integration(integration) do
-    if Integrations.due_for_sync?(integration, 10) do
-      case Sync.sync_organization(integration) do
-        {:ok, result} ->
-          Logger.info("Sonar sync completed for org #{integration.organization_id}: #{inspect(result)}")
-          {:ok, result}
+  # Individual sync: loads integration by ID and syncs.
+  def perform(%Oban.Job{args: %{"integration_id" => id}}) do
+    case Integrations.get_integration_by_id(id) do
+      {:ok, integration} ->
+        sync_integration(integration)
 
-        {:error, reason} ->
-          Logger.error("Sonar sync failed for org #{integration.organization_id}: #{inspect(reason)}")
-          {:error, reason}
-      end
-    else
-      :skipped
+      {:error, :not_found} ->
+        Logger.warning("Sonar sync: integration #{id} not found, skipping")
+        {:error, :not_found}
     end
   end
+
+  defp sync_integration(integration) do
+    case Sync.sync_organization(integration) do
+      {:ok, result} ->
+        Logger.info("Sonar sync completed for org #{integration.organization_id}: #{inspect(result)}")
+        :ok
+
+      {:error, reason} ->
+        Logger.error("Sonar sync failed for org #{integration.organization_id}: #{inspect(reason)}")
+        # Permanent failures (auth, bad config) should not be retried;
+        # transient ones (5xx, timeouts, rate limits) should.
+        if SyncErrors.transient?(reason), do: {:error, reason}, else: :ok
+    end
+  end
+
+  @doc false
+  def due_for_sync?(integration), do: Integrations.due_for_sync?(integration, @default_interval_minutes)
 end
diff --git a/lib/towerops/workers/splynx_sync_worker.ex b/lib/towerops/workers/splynx_sync_worker.ex
index c75b9d31..25734056 100644
--- a/lib/towerops/workers/splynx_sync_worker.ex
+++ b/lib/towerops/workers/splynx_sync_worker.ex
@@ -1,44 +1,82 @@
 defmodule Towerops.Workers.SplynxSyncWorker do
   @moduledoc """
-  Oban cron worker that syncs Splynx data for all enabled integrations.
+  Oban worker that syncs Splynx data for enabled integrations.
+
+  The cron job (no args) runs every 5 minutes and dispatches individual
+  per-integration sync jobs staggered across the 300-second window using
+  `PollingOffset.calculate_offset/2`. This prevents a thundering herd
+  against Splynx APIs and stops one slow tenant from blocking the
+  maintenance queue while every other integration waits in line.
+
+  Individual jobs receive `%{"integration_id" => id}` and perform the
+  actual sync via `Splynx.Sync.sync_organization/1`.
   """
   use Oban.Worker, queue: :maintenance
 
   alias Towerops.Integrations
   alias Towerops.Splynx.Sync
+  alias Towerops.Workers.PollingOffset
+  alias Towerops.Workers.SyncErrors
 
   require Logger
 
+  @provider "splynx"
+  @default_interval_minutes 10
+  @window_seconds 300
+
+  # Cron dispatcher: enqueues staggered individual sync jobs.
   @impl Oban.Worker
-  def perform(%Oban.Job{}) do
-    integrations = Integrations.list_enabled_integrations("splynx")
+  def perform(%Oban.Job{args: args}) when args == %{} do
+    integrations = Integrations.list_enabled_integrations(@provider)
 
-    results = Enum.map(integrations, &sync_integration/1)
+    eligible = Enum.filter(integrations, &due_for_sync?/1)
 
-    synced = Enum.count(results, &match?({:ok, _}, &1))
-    failed = Enum.count(results, &match?({:error, _}, &1))
-    skipped = Enum.count(results, &(&1 == :skipped))
+    now = DateTime.utc_now()
 
-    if synced > 0 or failed > 0 do
-      Logger.info("Splynx sync batch: #{synced} synced, #{failed} failed, #{skipped} skipped")
+    enqueued =
+      Enum.count(eligible, fn integration ->
+        offset = PollingOffset.calculate_offset(integration.organization_id, @window_seconds)
+        scheduled_at = DateTime.add(now, offset, :second)
+
+        case %{"integration_id" => integration.id}
+             |> new(scheduled_at: scheduled_at, unique: [period: @window_seconds])
+             |> Oban.insert() do
+          {:ok, _job} -> true
+          {:error, _reason} -> false
+        end
+      end)
+
+    if enqueued > 0 do
+      Logger.info("Splynx sync: dispatched #{enqueued} jobs across #{@window_seconds}s window")
     end
 
     :ok
   end
 
-  defp sync_integration(integration) do
-    if Integrations.due_for_sync?(integration, 10) do
-      case Sync.sync_organization(integration) do
-        {:ok, result} ->
-          Logger.info("Splynx sync completed for org #{integration.organization_id}: #{inspect(result)}")
-          {:ok, result}
+  # Individual sync: loads integration by ID and syncs.
+  def perform(%Oban.Job{args: %{"integration_id" => id}}) do
+    case Integrations.get_integration_by_id(id) do
+      {:ok, integration} ->
+        sync_integration(integration)
 
-        {:error, reason} ->
-          Logger.error("Splynx sync failed for org #{integration.organization_id}: #{inspect(reason)}")
-          {:error, reason}
-      end
-    else
-      :skipped
+      {:error, :not_found} ->
+        Logger.warning("Splynx sync: integration #{id} not found, skipping")
+        {:error, :not_found}
     end
   end
+
+  defp sync_integration(integration) do
+    case Sync.sync_organization(integration) do
+      {:ok, result} ->
+        Logger.info("Splynx sync completed for org #{integration.organization_id}: #{inspect(result)}")
+        :ok
+
+      {:error, reason} ->
+        Logger.error("Splynx sync failed for org #{integration.organization_id}: #{inspect(reason)}")
+        if SyncErrors.transient?(reason), do: {:error, reason}, else: :ok
+    end
+  end
+
+  @doc false
+  def due_for_sync?(integration), do: Integrations.due_for_sync?(integration, @default_interval_minutes)
 end
diff --git a/lib/towerops/workers/visp_sync_worker.ex b/lib/towerops/workers/visp_sync_worker.ex
index 3e7588c4..bdb8b227 100644
--- a/lib/towerops/workers/visp_sync_worker.ex
+++ b/lib/towerops/workers/visp_sync_worker.ex
@@ -1,44 +1,82 @@
 defmodule Towerops.Workers.VispSyncWorker do
   @moduledoc """
-  Oban cron worker that syncs VISP data for all enabled integrations.
+  Oban worker that syncs VISP data for enabled integrations.
+
+  The cron job (no args) runs every 5 minutes and dispatches individual
+  per-integration sync jobs staggered across the 300-second window using
+  `PollingOffset.calculate_offset/2`. This prevents a thundering herd
+  against VISP APIs and stops one slow tenant from blocking the
+  maintenance queue while every other integration waits in line.
+
+  Individual jobs receive `%{"integration_id" => id}` and perform the
+  actual sync via `Visp.Sync.sync_organization/1`.
   """
   use Oban.Worker, queue: :maintenance
 
   alias Towerops.Integrations
   alias Towerops.Visp.Sync
+  alias Towerops.Workers.PollingOffset
+  alias Towerops.Workers.SyncErrors
 
   require Logger
 
+  @provider "visp"
+  @default_interval_minutes 10
+  @window_seconds 300
+
+  # Cron dispatcher: enqueues staggered individual sync jobs.
   @impl Oban.Worker
-  def perform(%Oban.Job{}) do
-    integrations = Integrations.list_enabled_integrations("visp")
+  def perform(%Oban.Job{args: args}) when args == %{} do
+    integrations = Integrations.list_enabled_integrations(@provider)
 
-    results = Enum.map(integrations, &sync_integration/1)
+    eligible = Enum.filter(integrations, &due_for_sync?/1)
 
-    synced = Enum.count(results, &match?({:ok, _}, &1))
-    failed = Enum.count(results, &match?({:error, _}, &1))
-    skipped = Enum.count(results, &(&1 == :skipped))
+    now = DateTime.utc_now()
 
-    if synced > 0 or failed > 0 do
-      Logger.info("VISP sync batch: #{synced} synced, #{failed} failed, #{skipped} skipped")
+    enqueued =
+      Enum.count(eligible, fn integration ->
+        offset = PollingOffset.calculate_offset(integration.organization_id, @window_seconds)
+        scheduled_at = DateTime.add(now, offset, :second)
+
+        case %{"integration_id" => integration.id}
+             |> new(scheduled_at: scheduled_at, unique: [period: @window_seconds])
+             |> Oban.insert() do
+          {:ok, _job} -> true
+          {:error, _reason} -> false
+        end
+      end)
+
+    if enqueued > 0 do
+      Logger.info("VISP sync: dispatched #{enqueued} jobs across #{@window_seconds}s window")
     end
 
     :ok
   end
 
-  defp sync_integration(integration) do
-    if Integrations.due_for_sync?(integration, 10) do
-      case Sync.sync_organization(integration) do
-        {:ok, result} ->
-          Logger.info("VISP sync completed for org #{integration.organization_id}: #{inspect(result)}")
-          {:ok, result}
+  # Individual sync: loads integration by ID and syncs.
+  def perform(%Oban.Job{args: %{"integration_id" => id}}) do
+    case Integrations.get_integration_by_id(id) do
+      {:ok, integration} ->
+        sync_integration(integration)
 
-        {:error, reason} ->
-          Logger.error("VISP sync failed for org #{integration.organization_id}: #{inspect(reason)}")
-          {:error, reason}
-      end
-    else
-      :skipped
+      {:error, :not_found} ->
+        Logger.warning("VISP sync: integration #{id} not found, skipping")
+        {:error, :not_found}
     end
   end
+
+  defp sync_integration(integration) do
+    case Sync.sync_organization(integration) do
+      {:ok, result} ->
+        Logger.info("VISP sync completed for org #{integration.organization_id}: #{inspect(result)}")
+        :ok
+
+      {:error, reason} ->
+        Logger.error("VISP sync failed for org #{integration.organization_id}: #{inspect(reason)}")
+        if SyncErrors.transient?(reason), do: {:error, reason}, else: :ok
+    end
+  end
+
+  @doc false
+  def due_for_sync?(integration), do: Integrations.due_for_sync?(integration, @default_interval_minutes)
 end
diff --git a/test/towerops/workers/gaiia_sync_worker_test.exs b/test/towerops/workers/gaiia_sync_worker_test.exs
index 8383055b..303d12ba 100644
--- a/test/towerops/workers/gaiia_sync_worker_test.exs
+++ b/test/towerops/workers/gaiia_sync_worker_test.exs
@@ -1,5 +1,6 @@
 defmodule Towerops.Workers.GaiiaSyncWorkerTest do
-  use Towerops.DataCase, async: true
+  use Towerops.DataCase, async: false
+  use Oban.Testing, repo: Towerops.Repo
 
   import Towerops.AccountsFixtures
   import Towerops.IntegrationsFixtures
@@ -15,22 +16,42 @@ defmodule Towerops.Workers.GaiiaSyncWorkerTest do
     %{org: org, integration: integration}
   end
 
-  describe "perform/1" do
-    test "syncs enabled gaiia integrations", %{org: org} do
+  describe "perform/1 dispatcher (empty args)" do
+    test "enqueues a per-integration sync job for each eligible integration", %{integration: integration} do
+      assert :ok = perform_job(GaiiaSyncWorker, %{})
+
+      jobs = all_enqueued(worker: GaiiaSyncWorker)
+      assert length(jobs) == 1
+      [job] = jobs
+      assert job.args == %{"integration_id" => integration.id}
+    end
+
+    test "skips recently synced integrations", %{integration: integration} do
+      # Mark as recently synced — the dispatcher should now enqueue nothing.
+      Towerops.Integrations.update_sync_status(integration, "success")
+
+      assert :ok = perform_job(GaiiaSyncWorker, %{})
+      assert [] = all_enqueued(worker: GaiiaSyncWorker)
+    end
+  end
+
+  describe "perform/1 (with integration_id)" do
+    test "syncs a single enabled gaiia integration", %{org: org, integration: integration} do
       stub_all_endpoints()
 
-      assert :ok = GaiiaSyncWorker.perform(%Oban.Job{})
+      assert :ok = perform_job(GaiiaSyncWorker, %{"integration_id" => integration.id})
 
       # Verify data was synced
       assert length(Towerops.Gaiia.list_accounts(org.id)) == 1
     end
 
-    test "skips recently synced integrations", %{integration: integration} do
-      # Mark as recently synced
-      Towerops.Integrations.update_sync_status(integration, "success")
+    test "returns {:error, :not_found} for a missing integration" do
+      missing_id = Ecto.UUID.generate()
 
-      # Should skip since it was just synced
-      assert :ok = GaiiaSyncWorker.perform(%Oban.Job{})
+      ExUnit.CaptureLog.capture_log(fn ->
+        assert {:error, :not_found} =
+                 perform_job(GaiiaSyncWorker, %{"integration_id" => missing_id})
+      end)
     end
   end
 
diff --git a/test/towerops/workers/sync_worker_perform_test.exs b/test/towerops/workers/sync_worker_perform_test.exs
index 7358a7be..50f3dec2 100644
--- a/test/towerops/workers/sync_worker_perform_test.exs
+++ b/test/towerops/workers/sync_worker_perform_test.exs
@@ -37,18 +37,19 @@ defmodule Towerops.Workers.SyncWorkerPerformTest do
         conn |> Plug.Conn.put_status(401) |> Req.Test.json(%{"error" => "unauthorized"})
       end)
 
-      {:ok, _integration} =
+      {:ok, integration} =
         Integrations.create_integration(org.id, %{
           provider: "sonar",
           enabled: true,
           credentials: %{"instance_url" => "https://sonar.test", "api_token" => "bad"}
         })
 
-      assert :ok = perform_job(SonarSyncWorker, %{})
+      # Dispatcher only enqueues; run the per-integration job directly.
+      assert :ok = perform_job(SonarSyncWorker, %{"integration_id" => integration.id})
 
       # Status should now be flagged as failed.
-      {:ok, integration} = Integrations.get_integration(org.id, "sonar")
-      assert integration.last_sync_status == "failed"
+      {:ok, reloaded} = Integrations.get_integration(org.id, "sonar")
+      assert reloaded.last_sync_status == "failed"
     end
   end
 
@@ -58,7 +59,7 @@ defmodule Towerops.Workers.SyncWorkerPerformTest do
         conn |> Plug.Conn.put_status(401) |> Req.Test.json(%{"error" => "unauthorized"})
       end)
 
-      {:ok, _integration} =
+      {:ok, integration} =
         Integrations.create_integration(org.id, %{
           provider: "splynx",
           enabled: true,
@@ -69,10 +70,10 @@ defmodule Towerops.Workers.SyncWorkerPerformTest do
           }
         })
 
-      assert :ok = perform_job(SplynxSyncWorker, %{})
+      assert :ok = perform_job(SplynxSyncWorker, %{"integration_id" => integration.id})
 
-      {:ok, integration} = Integrations.get_integration(org.id, "splynx")
-      assert integration.last_sync_status == "failed"
+      {:ok, reloaded} = Integrations.get_integration(org.id, "splynx")
+      assert reloaded.last_sync_status == "failed"
     end
   end
 
@@ -82,17 +83,18 @@ defmodule Towerops.Workers.SyncWorkerPerformTest do
         conn |> Plug.Conn.put_status(500) |> Req.Test.json(%{"error" => "boom"})
       end)
 
-      {:ok, _integration} =
+      {:ok, integration} =
         Integrations.create_integration(org.id, %{
           provider: "visp",
           enabled: true,
           credentials: %{"api_key" => "bad"}
         })
 
-      assert :ok = perform_job(VispSyncWorker, %{})
+      # 500 is transient — perform returns {:error, _} so Oban will retry.
+      assert {:error, _} = perform_job(VispSyncWorker, %{"integration_id" => integration.id})
 
-      {:ok, integration} = Integrations.get_integration(org.id, "visp")
-      assert integration.last_sync_status == "failed"
+      {:ok, reloaded} = Integrations.get_integration(org.id, "visp")
+      assert reloaded.last_sync_status == "failed"
     end
   end
 
@@ -102,7 +104,7 @@ defmodule Towerops.Workers.SyncWorkerPerformTest do
         conn |> Plug.Conn.put_status(401) |> Req.Test.json(%{"detail" => "unauthorized"})
       end)
 
-      {:ok, _integration} =
+      {:ok, integration} =
         Integrations.create_integration(org.id, %{
           provider: "netbox",
           enabled: true,
@@ -113,21 +115,21 @@ defmodule Towerops.Workers.SyncWorkerPerformTest do
           }
         })
 
-      assert :ok = perform_job(NetBoxSyncWorker, %{})
+      assert :ok = perform_job(NetBoxSyncWorker, %{"integration_id" => integration.id})
 
-      {:ok, integration} = Integrations.get_integration(org.id, "netbox")
+      {:ok, reloaded} = Integrations.get_integration(org.id, "netbox")
       # NetBox sync may report status as success/failed depending on exact
       # error path; just confirm the worker ran without raising and the
       # integration is reachable. Either way, the perform/1 body executed.
-      assert is_binary(integration.last_sync_status) or is_nil(integration.last_sync_status)
+      assert is_binary(reloaded.last_sync_status) or is_nil(reloaded.last_sync_status)
     end
   end
 
-  describe "all sync workers tolerate `should_sync?` returning false" do
-    test "skipped integrations don't update last_sync_status", %{org: org} do
+  describe "dispatchers skip recently-synced integrations" do
+    test "dispatcher does not enqueue a job when the interval hasn't elapsed", %{org: org} do
       now = DateTime.truncate(DateTime.utc_now(), :second)
 
-      # Recent last_synced_at + default 10-min interval → should_sync? returns false.
+      # Recent last_synced_at + default 10-min interval → due_for_sync? returns false.
       {:ok, integration} =
         Integrations.create_integration(org.id, %{
           provider: "sonar",
@@ -139,8 +141,10 @@ defmodule Towerops.Workers.SyncWorkerPerformTest do
 
       assert :ok = perform_job(SonarSyncWorker, %{})
 
+      # Nothing got queued, and the integration row is untouched.
+      assert [] = all_enqueued(worker: SonarSyncWorker)
+
       {:ok, reloaded} = Integrations.get_integration(org.id, "sonar")
-      # Status is unchanged because the worker skipped this one.
       assert reloaded.last_sync_status == "success"
       assert reloaded.last_synced_at == integration.last_synced_at
     end