feat: Add Pause and Resume functionality for campaigns (#1204)

* feat(backend): Add pause and resume mutations for campaigns

- Pausing a campaign stops any new target execution and disables
retries, while continuing to listen for and process device updates
for already in-progress operations.
- Resuming a campaign restores retry timers and continues the rollout
from the current state (same counters, same slots),
fetching new targets as capacity allows.
- Add missing tests for deployment upgrade campaigns.

Adds the ability to pause and resume campaigns from the Campaign
Details page. When a campaign is paused, no new targets will be
assigned, but in-progress targets will continue until completion.

Closes #277

Signed-off-by: Omar <omar.brbutovic@secomind.com>

Author: OmarBrbutovic

backend/lib/edgehog/campaigns/campaign/campaign.ex

@@ -22,7 +22,8 @@ defmodule Edgehog.Campaigns.Campaign do
   @moduledoc false
   use Edgehog.MultitenantResource,
     domain: Edgehog.Campaigns,
-    extensions: [AshGraphql.Resource]
+    extensions: [AshGraphql.Resource],
+    notifiers: [Ash.Notifier.PubSub]
 
   alias Edgehog.Campaigns.Campaign.Changes
   alias Edgehog.Campaigns.Campaign.Validations
@@ -44,7 +45,7 @@ defmodule Edgehog.Campaigns.Campaign do
     read :read_all_resumable do
       multitenancy :allow_global
       pagination keyset?: true
-      filter expr(status in [:idle, :in_progress])
+      filter expr(status in [:idle, :in_progress, :pausing])
     end
 
     read :update_campaign do
@@ -121,6 +122,24 @@ defmodule Edgehog.Campaigns.Campaign do
       change set_attribute(:outcome, :success)
     end
 
+    update :mark_as_paused do
+      change set_attribute(:status, :paused)
+    end
+
+    update :pause do
+      require_atomic? false
+
+      validate {Validations.ValidateStatus, operation: :pause}
+      change set_attribute(:status, :pausing)
+    end
+
+    update :resume do
+      require_atomic? false
+
+      validate {Validations.ValidateStatus, operation: :resume}
+      change Changes.StartExecution
+    end
+
     destroy :destroy do
       description "Deletes a Campaign"
       primary? true
@@ -206,6 +225,13 @@ defmodule Edgehog.Campaigns.Campaign do
     end
   end
 
+  pub_sub do
+    prefix "campaigns"
+    module EdgehogWeb.Endpoint
+
+    publish :pause, [[:id, "*"]]
+  end
+
   postgres do
     table "campaigns"
 

backend/lib/edgehog/campaigns/campaign/changes/start_execution.ex

@@ -0,0 +1,44 @@
+#
+# This file is part of Edgehog.
+#
+# Copyright 2026 SECO Mind Srl
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# SPDX-License-Identifier: Apache-2.0
+#
+
+defmodule Edgehog.Campaigns.Campaign.Changes.StartExecution do
+  @moduledoc false
+
+  use Ash.Resource.Change
+
+  alias Edgehog.Campaigns.ExecutorSupervisor
+
+  @impl Ash.Resource.Change
+  def change(changeset, _opts, _context) do
+    changeset
+    |> Ash.Changeset.change_attribute(:status, :in_progress)
+    |> Ash.Changeset.after_transaction(fn _changeset, result ->
+      start_campaign_executor(result)
+    end)
+  end
+
+  defp start_campaign_executor({:ok, campaign} = _transaction_result) do
+    _pid = ExecutorSupervisor.start_executor!(campaign)
+
+    {:ok, campaign}
+  end
+
+  defp start_campaign_executor(transaction_result), do: transaction_result
+end

backend/lib/edgehog/campaigns/campaign/validations/validate_status.ex

@@ -0,0 +1,67 @@
+#
+# This file is part of Edgehog.
+#
+# Copyright 2026 SECO Mind Srl
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# SPDX-License-Identifier: Apache-2.0
+#
+
+defmodule Edgehog.Campaigns.Campaign.Validations.ValidateStatus do
+  @moduledoc """
+  Validates that a campaign is in the expected status for an operation.
+
+  ## Options
+  - `operation` - The operation being performed, used in error messages (required)
+  """
+  use Ash.Resource.Validation
+
+  alias Ash.Error.Changes
+
+  @impl Ash.Resource.Validation
+  def validate(changeset, opts, _context) do
+    operation = Keyword.fetch!(opts, :operation)
+    status = Ash.Changeset.get_attribute(changeset, :status)
+
+    validate_transition(status, operation)
+  end
+
+  defp validate_transition(status, :pause) do
+    case status do
+      :in_progress ->
+        :ok
+
+      _other ->
+        {:error,
+         Changes.InvalidAttribute.exception(
+           field: :status,
+           message: "Cannot pause campaign. Campaign must be in progress (current status: #{status})"
+         )}
+    end
+  end
+
+  defp validate_transition(status, :resume) do
+    case status do
+      :paused ->
+        :ok
+
+      _other ->
+        {:error,
+         Changes.InvalidAttribute.exception(
+           field: :status,
+           message: "Cannot resume campaign. Campaign must be paused (current status: #{status})"
+         )}
+    end
+  end
+end

backend/lib/edgehog/campaigns/campaign_mechanism/core.ex

@@ -32,6 +32,7 @@ defprotocol Edgehog.Campaigns.CampaignMechanism.Core do
   def mark_campaign_in_progress!(mechanism, campaign, now \\ DateTime.utc_now())
   def mark_campaign_as_failed!(mechanism, campaign, now \\ DateTime.utc_now())
   def mark_campaign_as_successful!(mechanism, campaign, now \\ DateTime.utc_now())
+  def mark_campaign_as_paused!(mechanism, campaign)
   def get_campaign_status(mechanism, campaign)
   def get_target_count(mechanism, tenant_id, campaign_id)
   def get_failed_target_count(mechanism, tenant_id, campaign_id)
@@ -139,19 +140,33 @@ defimpl Edgehog.Campaigns.CampaignMechanism.Core, for: Any do
     Campaigns.mark_campaign_successful!(campaign, %{completion_timestamp: now})
   end
 
+  @doc """
+  Marks a campaign as paused.
+
+  ## Parameters
+    - mechanism: The campaign mechanism (unused in default implementation).
+    - campaign: The campaign struct.
+
+  ## Returns
+    - The updated campaign struct marked as paused.
+  """
+  def mark_campaign_as_paused!(_mechanism, campaign) do
+    Campaigns.mark_campaign_paused!(campaign)
+  end
+
   # Campaign Data
 
   @doc """
   Return the persisted campaign status.
 
-  Expected values are `:idle`, `:in_progress` or `:finished`.
+  Expected values are `:idle`, `:in_progress`, `:pausing`, `:paused` or `:finished`.
 
   ## Parameters
     - mechanism: The campaign mechanism (unused in default implementation).
     - campaign: The campaign struct.
 
   ## Returns
-    - The campaign status atom (`:idle`, `:in_progress`, or `:finished`).
+    - The campaign status atom.
   """
   def get_campaign_status(_mechanism, campaign), do: campaign.status
 

backend/lib/edgehog/campaigns/campaign_mechanism/deployment_delete/core.ex

@@ -288,6 +288,9 @@ defimpl Edgehog.Campaigns.CampaignMechanism.Core,
   defdelegate mark_campaign_as_successful!(mechanism, campaign, now \\ DateTime.utc_now()),
     to: Any
 
+  defdelegate mark_campaign_as_paused!(mechanism, campaign),
+    to: Any
+
   defdelegate get_campaign_status(mechanism, campaign),
     to: Any
 

backend/lib/edgehog/campaigns/campaign_mechanism/deployment_delete/executor.ex

@@ -32,15 +32,23 @@ defmodule Edgehog.Campaigns.CampaignMechanism.DeploymentDelete.Executor do
     {:next_state, :initialization, data, internal_event(:init_data)}
   end
 
+  # Valid states from which pausing is allowed
+  @pauseable_states [
+    :execution,
+    :wait_for_available_slot,
+    :wait_for_target,
+    :wait_for_campaign_completion
+  ]
+
   # Common event handling
 
   # Note that external (e.g. :info) and timeout events are always handled after the internal
   # events enqueued with the :next_event action. This means that we can be sure an :info event
   # or a timeout won't be handled, e.g., between a rollout and the handling of its error
   @impl LazyBatch
-  def handle_info(%Phoenix.Socket.Broadcast{} = notification, _state, data) do
+  def handle_info(%Phoenix.Socket.Broadcast{} = notification, state, data) do
     case notification.payload.action.type do
-      :update -> handle_update(notification, data)
+      :update -> handle_update(notification, state, data)
       :destroy -> handle_destroy(notification, data)
       _ -> :keep_state_and_data
     end
@@ -51,9 +59,10 @@ defmodule Edgehog.Campaigns.CampaignMechanism.DeploymentDelete.Executor do
     :keep_state_and_data
   end
 
-  defp handle_update(notification, data) do
+  defp handle_update(notification, state, data) do
     case notification.payload.action.name do
       :mark_as_timed_out -> handle_mark_as_timed_out(notification, data)
+      :pause -> handle_mark_as_paused(state, data)
       _ -> :keep_state_and_data
     end
   end
@@ -92,4 +101,13 @@ defmodule Edgehog.Campaigns.CampaignMechanism.DeploymentDelete.Executor do
 
     {:keep_state_and_data, actions}
   end
+
+  defp handle_mark_as_paused(state, data) when state in @pauseable_states do
+    {:next_state, :wait_for_campaign_paused, data, []}
+  end
+
+  defp handle_mark_as_paused(_state, _data) do
+    # Ignore pause requests in non-pauseable states (terminal states, already pausing, etc.)
+    :keep_state_and_data
+  end
 end

backend/lib/edgehog/campaigns/campaign_mechanism/deployment_deploy/core.ex

@@ -281,6 +281,9 @@ defimpl Edgehog.Campaigns.CampaignMechanism.Core,
   defdelegate mark_campaign_as_successful!(mechanism, campaign, now \\ DateTime.utc_now()),
     to: Any
 
+  defdelegate mark_campaign_as_paused!(mechanism, campaign),
+    to: Any
+
   defdelegate get_campaign_status(mechanism, campaign),
     to: Any
 

backend/lib/edgehog/campaigns/campaign_mechanism/deployment_deploy/executor.ex

@@ -32,15 +32,23 @@ defmodule Edgehog.Campaigns.CampaignMechanism.DeploymentDeploy.Executor do
     {:next_state, :initialization, data, internal_event(:init_data)}
   end
 
-  # Common event handling
-
   # Note that external (e.g. :info) and timeout events are always handled after the internal
   # events enqueued with the :next_event action. This means that we can be sure an :info event
   # or a timeout won't be handled, e.g., between a rollout and the handling of its error
+  # Valid states from which pausing is allowed
+  @pauseable_states [
+    :execution,
+    :wait_for_available_slot,
+    :wait_for_target,
+    :wait_for_campaign_completion
+  ]
+
+  # Common event handling
+
   @impl LazyBatch
-  def handle_info(%Phoenix.Socket.Broadcast{} = notification, _state, data) do
+  def handle_info(%Phoenix.Socket.Broadcast{} = notification, state, data) do
     case notification.payload.action.type do
-      :update -> handle_update(notification, data)
+      :update -> handle_update(notification, state, data)
       _ -> :keep_state_and_data
     end
   end
@@ -50,10 +58,11 @@ defmodule Edgehog.Campaigns.CampaignMechanism.DeploymentDeploy.Executor do
     :keep_state_and_data
   end
 
-  defp handle_update(notification, data) do
+  defp handle_update(notification, state, data) do
     case notification.payload.action.name do
       :maybe_run_ready_actions -> handle_maybe_run_ready_actions(notification, data)
       :mark_as_timed_out -> handle_mark_as_timed_out(notification, data)
+      :pause -> handle_mark_as_paused(state, data)
       _ -> :keep_state_and_data
     end
   end
@@ -94,4 +103,13 @@ defmodule Edgehog.Campaigns.CampaignMechanism.DeploymentDeploy.Executor do
 
     {:keep_state_and_data, actions}
   end
+
+  defp handle_mark_as_paused(state, data) when state in @pauseable_states do
+    {:next_state, :wait_for_campaign_paused, data, []}
+  end
+
+  defp handle_mark_as_paused(_state, _data) do
+    # Ignore pause requests in non-pauseable states (terminal states, already pausing, etc.)
+    :keep_state_and_data
+  end
 end

backend/lib/edgehog/campaigns/campaign_mechanism/deployment_start/core.ex

@@ -289,6 +289,9 @@ defimpl Edgehog.Campaigns.CampaignMechanism.Core,
   defdelegate mark_campaign_as_successful!(mechanism, campaign, now \\ DateTime.utc_now()),
     to: Any
 
+  defdelegate mark_campaign_as_paused!(mechanism, campaign),
+    to: Any
+
   defdelegate get_campaign_status(mechanism, campaign),
     to: Any