fix(zebra): replace Task.async with Wormhole to fix Cachex issue (#612)

## 📝 Description

Replaced `Task.async/await` with the `Wormhole` library in
`Zebra.Workers.Scheduler.Org` to resolve Cachex blocking issues.

When using `Task.async/await` inside `Cachex.fetch/3`, if the fallback
function crashed or timed out, subsequent calls would block
indefinitely, as described in [the linked Cachex
issue](whitfin/cachex#220).

By switching to `Wormhole.capture/4`, the call is executed in a
supervised process with timeout handling and proper cleanup, preventing
Cachex lockups.

Added tests for error handling, timeouts, and caching behavior.

More details in [this
task](renderedtext/tasks#7605).

## ✅ Checklist
- [x] I have tested this change
- [x] ~This change requires documentation update~ N/A

Author: dexyk

zebra/lib/zebra/workers/scheduler/org.ex

@@ -18,23 +18,27 @@ defmodule Zebra.Workers.Scheduler.Org do
   """
   def load(org_id) do
     Zebra.Cache.fetch!("quotas-#{org_id}-v3", @cache_timeout, fn ->
-      cold_load(org_id)
+      result =
+        Wormhole.capture(__MODULE__, :fetch_org, [org_id],
+          timeout: 10_500,
+          stacktrace: true,
+          skip_log: true
+        )
+
+      case result do
+        {:ok, {:ok, org}} ->
+          {:commit, {:ok, new(org, org_id)}}
+
+        {:ok, error} ->
+          {:ignore, error}
+
+        error ->
+          {:ignore, error}
+      end
     end)
   end
 
-  defp cold_load(org_id) do
-    find_org = Task.async(fn -> fetch_org(org_id) end)
-
-    case Task.await(find_org) do
-      {:ok, org} ->
-        {:ok, new(org, org_id)}
-
-      e ->
-        e
-    end
-  end
-
-  defp fetch_org(org_id) do
+  def fetch_org(org_id) do
     alias InternalApi.Organization.DescribeRequest, as: Request
     alias InternalApi.Organization.OrganizationService.Stub
 

zebra/test/zebra/workers/scheduler/org_test.exs

@@ -0,0 +1,90 @@
+defmodule Zebra.Workers.Scheduler.OrgTest do
+  use Zebra.DataCase
+  alias Zebra.Workers.Scheduler.Org
+
+  describe "load with error handling" do
+    @org_id Ecto.UUID.generate()
+
+    setup do
+      # Reset the cache before each test
+      Cachex.clear(:zebra_cache)
+      :ok
+    end
+
+    test "returns data without caching when error is returned" do
+      # First set up a failure
+      alias Support.FakeServers.OrganizationApi, as: OrgApi
+
+      GrpcMock.stub(OrgApi, :describe, fn _, _ ->
+        {:error, "Connection error"}
+      end)
+
+      # Load should return error and not cache it
+      assert {:error, _} = Org.load(@org_id)
+
+      # Now set up a success response
+      GrpcMock.stub(OrgApi, :describe, fn _, _ ->
+        InternalApi.Organization.DescribeResponse.new(
+          status:
+            InternalApi.ResponseStatus.new(code: InternalApi.ResponseStatus.Code.value(:OK)),
+          organization: InternalApi.Organization.Organization.new(org_username: "testing-org")
+        )
+      end)
+
+      # It should now succeed because the error wasn't cached
+      assert {:ok, _} = Org.load(@org_id)
+    end
+
+    test "properly handles timeouts and doesn't cache them" do
+      # First set up a timeout scenario using Wormhole
+      alias Support.FakeServers.OrganizationApi, as: OrgApi
+
+      GrpcMock.stub(OrgApi, :describe, fn _, _ ->
+        # Sleep to simulate timeout longer than Wormhole's timeout
+        Process.sleep(15_000)
+        {:ok, "This shouldn't be returned"}
+      end)
+
+      # Load should return error due to timeout and not cache it
+      assert {:error, _} = Org.load(@org_id)
+
+      # Now set up a success response
+      GrpcMock.stub(OrgApi, :describe, fn _, _ ->
+        InternalApi.Organization.DescribeResponse.new(
+          status:
+            InternalApi.ResponseStatus.new(code: InternalApi.ResponseStatus.Code.value(:OK)),
+          organization: InternalApi.Organization.Organization.new(org_username: "testing-org")
+        )
+      end)
+
+      # It should now succeed because the timeout wasn't cached
+      assert {:ok, _} = Org.load(@org_id)
+    end
+
+    test "caches successful responses" do
+      # Set up a success response
+      alias Support.FakeServers.OrganizationApi, as: OrgApi
+
+      GrpcMock.stub(OrgApi, :describe, fn _, _ ->
+        InternalApi.Organization.DescribeResponse.new(
+          status:
+            InternalApi.ResponseStatus.new(code: InternalApi.ResponseStatus.Code.value(:OK)),
+          organization: InternalApi.Organization.Organization.new(org_username: "testing-org")
+        )
+      end)
+
+      # First call should succeed
+      assert {:ok, org} = Org.load(@org_id)
+      assert org.username == "testing-org"
+
+      # Now set up a failure, which shouldn't be used because we'll use the cached value
+      GrpcMock.stub(OrgApi, :describe, fn _, _ ->
+        {:error, "Connection error"}
+      end)
+
+      # It should still succeed because we're using the cached value
+      assert {:ok, org} = Org.load(@org_id)
+      assert org.username == "testing-org"
+    end
+  end
+end