fix(periodic_scheduler): handle partial unsuspend failures

Author: dexyk

periodic_scheduler/scheduler/docker-compose.yml

@@ -1,6 +1,6 @@
 services:
   app:
-    build: 
+    build:
       context: ../..
       dockerfile: periodic_scheduler/scheduler/Dockerfile
       target: dev
@@ -26,6 +26,8 @@ services:
     tty: true
     volumes:
       - ../:/app:delegated
+      - /app/scheduler/_build
+      - /app/scheduler/deps
     working_dir: "/app/scheduler"
 
   ciapp:
@@ -39,7 +41,7 @@ services:
       SSH_AUTH_SOCK: ${SSH_AUTH_SOCK}
     depends_on:
       postgres:
-        condition: service_healthy      
+        condition: service_healthy
       rabbitmq:
         condition: service_healthy
     profiles: ["ci"]

periodic_scheduler/scheduler/lib/scheduler/events_consumers/org_unblocked.ex

@@ -39,7 +39,12 @@ defmodule Scheduler.EventsConsumers.OrgUnblocked do
   defp unsuspend_batch(org_id, batch_no) do
     with {:ok, periodics} <- PeriodicsQueries.get_all_from_org(org_id, batch_no),
          {:periodics_found, true} <- {:periodics_found, length(periodics) > 0},
-         {:ok, _periodics} <- unsuspend_periodics(periodics) do
+         {:ok, %{failed: failed}} <- unsuspend_periodics(periodics) do
+      if failed != [] do
+        failed
+        |> LT.warn("Failed to unsuspend some periodics for organization #{org_id}")
+      end
+
       unsuspend_batch(org_id, batch_no + 1)
     else
       {:periodics_found, false} ->
@@ -51,13 +56,27 @@ defmodule Scheduler.EventsConsumers.OrgUnblocked do
   end
 
   defp unsuspend_periodics(periodics) do
-    periodics
-    |> Enum.reduce_while({:ok, []}, fn periodic, {:ok, results} ->
-      case unsuspend_periodic(periodic) do
-        {:ok, periodic} -> {:cont, {:ok, results ++ [periodic]}}
-        error -> {:halt, error}
-      end
-    end)
+    result =
+      Enum.reduce(periodics, %{unsuspended: [], failed: []}, fn periodic, acc ->
+        case unsuspend_periodic(periodic) do
+          {:ok, periodic} ->
+            %{acc | unsuspended: [periodic | acc.unsuspended]}
+
+          {:error, reason} ->
+            failed_entry = %{id: periodic.id, reason: reason}
+            %{acc | failed: [failed_entry | acc.failed]}
+
+          other ->
+            failed_entry = %{id: periodic.id, reason: other}
+            %{acc | failed: [failed_entry | acc.failed]}
+        end
+      end)
+
+    {:ok,
+     %{
+       unsuspended: Enum.reverse(result.unsuspended),
+       failed: Enum.reverse(result.failed)
+     }}
   end
 
   defp unsuspend_periodic(periodic) do

periodic_scheduler/scheduler/test/events_consumers/org_unblocked_test.exs

@@ -1,27 +1,50 @@
 defmodule Scheduler.EventsConsumers.OrgUnblocked.Test do
   use ExUnit.Case
 
+  import Ecto.Query, only: [from: 2]
+
   alias InternalApi.Organization.OrganizationUnblocked
   alias Scheduler.Periodics.Model.PeriodicsQueries
+  alias Scheduler.Periodics.Model.Periodics
+  alias Scheduler.PeriodicsRepo
   alias Scheduler.EventsConsumers.OrgUnblocked
   alias Scheduler.Workers.QuantumScheduler
   alias Scheduler.Actions
   alias Util.Proto
 
+  @grpc_port 50_057
+  @mocked_services [Test.MockFeatureService]
+
+  setup_all do
+    GRPC.Server.start(@mocked_services, @grpc_port)
+    {:ok, consumer_pid} = start_org_unblocked_consumer()
+
+    on_exit(fn ->
+      stop_org_unblocked_consumer(consumer_pid)
+      GRPC.Server.stop(@mocked_services)
+    end)
+
+    :ok
+  end
+
   setup do
     Test.Helpers.truncate_db()
 
     Test.Helpers.purge_queue("unblocked")
 
     ids_1 = Test.Helpers.seed_front_db()
     ids_2 = Test.Helpers.seed_front_db()
+    ids_3 = Test.Helpers.seed_front_db()
 
     start_supervised!(QuantumScheduler)
 
     System.put_env("GITHUB_APP_ID", "client_id")
     System.put_env("GITHUB_SECRET_ID", "client_secret")
 
-    {:ok, %{ids_1: ids_1, ids_2: ids_2}}
+    reset_mock_feature_service()
+    mock_feature_response("just_run")
+
+    {:ok, %{ids_1: ids_1, ids_2: ids_2, ids_3: ids_3}}
   end
 
   test "valid message received => all periodics from the org are unsuspended", ctx do
@@ -44,15 +67,40 @@ defmodule Scheduler.EventsConsumers.OrgUnblocked.Test do
     event = Proto.deep_new!(OrganizationUnblocked, %{org_id: ctx.ids_1.org_id})
     encoded = OrganizationUnblocked.encode(event)
 
-    assert {:ok, _pid} = OrgUnblocked.start_link()
-
     Tackle.publish(encoded, exchange_params())
 
     :timer.sleep(2_000)
 
     assert_only_periodics_from_unblocked_org_unsuspended(periodics, ctx.ids_1.org_id)
   end
 
+  test "invalid cron expression does not block unsuspending other periodics", ctx do
+    [{:ok, first_id}, {:ok, invalid_id}, {:ok, third_id}] =
+      1..3
+      |> Enum.map(&create_periodic(ctx.ids_3, &1))
+
+    invalidate_cron_expression(invalid_id)
+
+    event = Proto.deep_new!(OrganizationUnblocked, %{org_id: ctx.ids_3.org_id})
+    encoded = OrganizationUnblocked.encode(event)
+
+    Tackle.publish(encoded, exchange_params())
+
+    :timer.sleep(2_000)
+
+    assert {:ok, first_periodic} = PeriodicsQueries.get_by_id(first_id)
+    assert first_periodic.suspended == false
+    assert nil != first_id |> String.to_atom() |> QuantumScheduler.find_job()
+
+    assert {:ok, invalid_periodic} = PeriodicsQueries.get_by_id(invalid_id)
+    assert invalid_periodic.suspended == false
+    assert nil == invalid_id |> String.to_atom() |> QuantumScheduler.find_job()
+
+    assert {:ok, third_periodic} = PeriodicsQueries.get_by_id(third_id)
+    assert third_periodic.suspended == false
+    assert nil != third_id |> String.to_atom() |> QuantumScheduler.find_job()
+  end
+
   defp create_periodic(ids, ind) do
     {:ok, id} =
       %{
@@ -96,11 +144,59 @@ defmodule Scheduler.EventsConsumers.OrgUnblocked.Test do
     end)
   end
 
+  defp invalidate_cron_expression(id) do
+    from(p in Periodics, where: p.id == ^id)
+    |> PeriodicsRepo.update_all(set: [at: "invalid cron"])
+  end
+
   defp exchange_params() do
     %{
       url: System.get_env("RABBITMQ_URL"),
       exchange: "organization_exchange",
       routing_key: "unblocked"
     }
   end
+
+  defp start_org_unblocked_consumer do
+    case OrgUnblocked.start_link() do
+      {:ok, pid} ->
+        wait_for_org_unblocked_consumer()
+        {:ok, pid}
+
+      {:error, {:already_started, pid}} ->
+        wait_for_org_unblocked_consumer()
+        {:ok, pid}
+
+      error ->
+        error
+    end
+  end
+
+  defp wait_for_org_unblocked_consumer do
+    # Give Tackle time to register a default consumer before publishing
+    Process.sleep(200)
+  end
+
+  defp stop_org_unblocked_consumer(pid) when is_pid(pid) do
+    if Process.alive?(pid) do
+      GenServer.stop(pid)
+    else
+      :ok
+    end
+  end
+
+  defp stop_org_unblocked_consumer(_), do: :ok
+
+  defp reset_mock_feature_service() do
+    Cachex.clear(Elixir.Scheduler.FeatureHubProvider)
+
+    Application.put_env(
+      :scheduler,
+      :feature_api_grpc_endpoint,
+      "localhost:#{inspect(@grpc_port)}"
+    )
+  end
+
+  defp mock_feature_response(value),
+    do: Application.put_env(:scheduler, :mock_feature_service_response, value)
 end