fix: enqueue catalog sync after pool assignment changes

Queue a manual catalog sync after successful active Pool upstream assignment edits so model source metadata refreshes before the scheduler.

Keep the Pool mutation committed if the enqueue side effect fails and cover enqueue, no-enqueue, and drained metadata refresh behavior.

Author: masterkain

docs-site/src/content/docs/operators/jobs.mdx

@@ -63,8 +63,8 @@ The current worker groups are:
   <tbody>
     <tr>
       <td>Catalog sync</td>
-      <td>Refreshes model catalog metadata for active Pools.</td>
-      <td>Every 30 minutes, plus manual enqueue.</td>
+      <td>Refreshes model catalog metadata for active Pools. Successful upstream assignment edits on an active Pool enqueue an immediate asynchronous sync for that Pool so model source metadata can refresh before the next 30-minute run. Routing eligibility and in-flight work are unchanged.</td>
+      <td>Every 30 minutes, plus manual enqueue and successful active-Pool assignment edits.</td>
     </tr>
     <tr>
       <td>Pricing import</td>

docs-site/src/content/docs/operators/pools.mdx

@@ -239,6 +239,8 @@ The Upstreams step selects which active upstream accounts are available to the P
 
 Assign more than one upstream when you want load distribution, fallback, quota flexibility, or account maintenance without moving clients to a different API key. Keep an upstream unassigned when it should not receive work for that Pool.
 
+When a saved edit changes upstream assignments for an active Pool, Codex Pooler enqueues an immediate asynchronous catalog sync for that Pool so model source metadata can refresh without waiting for the 30-minute scheduler. The sync does not bypass runtime eligibility checks or change the routing state held by in-flight requests or already-open streams.
+
 ## API Keys Step
 
 The API keys step selects existing Pool API keys assigned to the Pool. Assigning a key lets that client credential use this Pool. Creating or rotating the raw key happens on the API keys page, where the raw secret is shown only once.

lib/codex_pooler/admin/pool_workflow.ex

@@ -3,14 +3,21 @@ defmodule CodexPooler.Admin.PoolWorkflow do
   Admin pool workflows that coordinate pool settings, upstream assignments, and API keys.
   """
 
+  require Logger
+
   alias CodexPooler.Access
   alias CodexPooler.Accounts.Scope
   alias CodexPooler.Events
+  alias CodexPooler.Jobs
   alias CodexPooler.Pools
   alias CodexPooler.Pools.Pool
   alias CodexPooler.Repo
   alias CodexPooler.Upstreams
 
+  @assignment_selection_keys ~w(upstream_identity_ids upstream_assignment_ids)
+  @catalog_sync_trigger_kind "manual"
+  @pool_active "active"
+
   @type access_error :: %{required(:code) => atom(), required(:message) => String.t()}
   @type pool_result :: {:ok, Pool.t()} | {:error, Ecto.Changeset.t() | access_error()}
 
@@ -36,6 +43,7 @@ defmodule CodexPooler.Admin.PoolWorkflow do
     end)
     |> normalize_transaction_result()
     |> maybe_broadcast_pool_workflow("pool_created")
+    |> maybe_enqueue_assignment_catalog_sync(attrs)
   end
 
   def create_pool_with_related_settings(_scope, _attrs),
@@ -53,6 +61,7 @@ defmodule CodexPooler.Admin.PoolWorkflow do
     end)
     |> normalize_transaction_result()
     |> maybe_broadcast_pool_workflow("pool_updated")
+    |> maybe_enqueue_assignment_catalog_sync(attrs)
   end
 
   def update_pool_with_related_settings(_scope, _pool_or_id, _attrs),
@@ -199,6 +208,46 @@ defmodule CodexPooler.Admin.PoolWorkflow do
   defp normalize_transaction_result({:ok, result}), do: {:ok, result}
   defp normalize_transaction_result({:error, reason}), do: {:error, reason}
 
+  @spec maybe_enqueue_assignment_catalog_sync(pool_result(), map()) :: pool_result()
+  defp maybe_enqueue_assignment_catalog_sync(
+         {:ok, %Pool{status: @pool_active} = pool} = result,
+         attrs
+       ) do
+    if assignment_selection_attrs?(attrs) do
+      case Jobs.enqueue_catalog_sync(pool, trigger_kind: @catalog_sync_trigger_kind) do
+        {:ok, _job} ->
+          result
+
+        {:error, reason} ->
+          log_catalog_sync_enqueue_failure(pool, reason)
+          result
+      end
+    else
+      result
+    end
+  end
+
+  defp maybe_enqueue_assignment_catalog_sync(result, _attrs), do: result
+
+  @spec assignment_selection_attrs?(map()) :: boolean()
+  defp assignment_selection_attrs?(attrs) when is_map(attrs) do
+    Enum.any?(@assignment_selection_keys, &Map.has_key?(attrs, &1))
+  end
+
+  @spec log_catalog_sync_enqueue_failure(Pool.t(), term()) :: :ok
+  defp log_catalog_sync_enqueue_failure(%Pool{} = pool, reason) do
+    reason = catalog_sync_enqueue_failure_code(reason)
+
+    Logger.warning(fn ->
+      "pool assignment catalog sync enqueue failed pool_id=#{pool.id} " <>
+        "trigger_kind=#{@catalog_sync_trigger_kind} reason=#{reason}"
+    end)
+  end
+
+  @spec catalog_sync_enqueue_failure_code(term()) :: String.t()
+  defp catalog_sync_enqueue_failure_code(%Ecto.Changeset{}), do: "invalid_job"
+  defp catalog_sync_enqueue_failure_code(reason) when is_atom(reason), do: Atom.to_string(reason)
+
   defp maybe_broadcast_pool_workflow({:ok, %Pool{} = pool} = result, reason) do
     Events.broadcast_pools(pool.id, reason, %{
       pool_id: pool.id,

test/codex_pooler/admin/pool_workflow_test.exs

@@ -3,7 +3,10 @@ defmodule CodexPooler.Admin.PoolWorkflowTest do
 
   alias CodexPooler.Accounts.Scope
   alias CodexPooler.Admin.PoolWorkflow
+  alias CodexPooler.Catalog
   alias CodexPooler.Events
+  alias CodexPooler.FakeUpstream
+  alias CodexPooler.Jobs.CatalogSyncWorker
   alias CodexPooler.Pools
   alias CodexPooler.Pools.Pool
   alias CodexPooler.Repo
@@ -14,6 +17,11 @@ defmodule CodexPooler.Admin.PoolWorkflowTest do
   import CodexPooler.AccountsFixtures
   import CodexPooler.PoolerFixtures
 
+  setup do
+    Repo.delete_all(Oban.Job)
+    :ok
+  end
+
   describe "pool workflow broadcasts" do
     test "broadcasts one pool event after coordinated creation commits" do
       %{user: owner} = bootstrap_owner_fixture(%{"email" => "owner@example.com"})
@@ -98,10 +106,169 @@ defmodule CodexPooler.Admin.PoolWorkflowTest do
       assert_receive {Events, event}
       assert event.pool_id == target_pool.id
       assert event.reason == "pool_updated"
+      assert_receive {Events, job_event}
+      assert job_event.pool_id == target_pool.id
+      assert job_event.reason == "job_status_updated"
+      assert job_event.payload["worker"] == "catalog_sync"
+      assert job_event.payload["status"] == "scheduled"
       refute_receive {Events, _event}
     end
   end
 
+  describe "pool assignment catalog sync enqueue" do
+    test "creation with upstream identity selection enqueues an immediate catalog sync" do
+      %{user: owner} = bootstrap_owner_fixture(%{"email" => "catalog-create-owner@example.com"})
+      scope = Scope.for_user(owner, ["instance_owner"])
+      identity = active_upstream_identity_fixture(%{chatgpt_account_id: "acct_catalog_create"})
+
+      assert {:ok, pool} =
+               PoolWorkflow.create_pool_with_related_settings(scope, %{
+                 "name" => "Catalog Create Sync",
+                 "routing_strategy" => "bridge_ring",
+                 "upstream_identity_ids" => [identity.id],
+                 "api_key_ids" => []
+               })
+
+      assert [job] = all_enqueued(worker: CatalogSyncWorker)
+      assert job.args == %{"pool_id" => pool.id, "trigger_kind" => "manual"}
+    end
+
+    test "active assignment edit enqueues one immediate catalog sync" do
+      %{user: owner} = bootstrap_owner_fixture(%{"email" => "catalog-edit-owner@example.com"})
+      scope = Scope.for_user(owner, ["instance_owner"])
+      pool = pool_fixture(%{slug: "catalog-edit-sync", name: "Catalog Edit Sync"})
+      %{assignment: assignment} = upstream_assignment_fixture(pool)
+
+      assert {:ok, updated_pool} =
+               PoolWorkflow.update_pool_with_related_settings(scope, pool, %{
+                 "name" => "Catalog Edit Sync Updated",
+                 "status" => "active",
+                 "routing_strategy" => "bridge_ring",
+                 "upstream_assignment_ids" => [assignment.id],
+                 "api_key_ids" => []
+               })
+
+      assert updated_pool.id == pool.id
+      assert [job] = all_enqueued(worker: CatalogSyncWorker)
+      assert job.args == %{"pool_id" => pool.id, "trigger_kind" => "manual"}
+    end
+
+    test "rolled back assignment edit enqueues no catalog sync" do
+      %{user: owner} = bootstrap_owner_fixture(%{"email" => "catalog-rollback-owner@example.com"})
+      scope = Scope.for_user(owner, ["instance_owner"])
+      missing_api_key_id = Ecto.UUID.generate()
+      identity = active_upstream_identity_fixture(%{chatgpt_account_id: "acct_catalog_rollback"})
+
+      assert {:error, %{message: "selected API keys are not available"}} =
+               PoolWorkflow.create_pool_with_related_settings(scope, %{
+                 "name" => "Catalog Rollback Sync",
+                 "routing_strategy" => "bridge_ring",
+                 "upstream_identity_ids" => [identity.id],
+                 "api_key_ids" => [missing_api_key_id]
+               })
+
+      refute Repo.get_by(Pool, slug: "catalog-rollback-sync")
+      assert [] = all_enqueued(worker: CatalogSyncWorker)
+    end
+
+    test "edit without assignment selection enqueues no catalog sync" do
+      %{user: owner} = bootstrap_owner_fixture(%{"email" => "catalog-noop-owner@example.com"})
+      scope = Scope.for_user(owner, ["instance_owner"])
+      pool = pool_fixture(%{slug: "catalog-noop-sync", name: "Catalog Noop Sync"})
+
+      assert {:ok, updated_pool} =
+               PoolWorkflow.update_pool_with_related_settings(scope, pool, %{
+                 "name" => "Catalog Noop Sync Updated",
+                 "status" => "active",
+                 "routing_strategy" => "bridge_ring",
+                 "api_key_ids" => []
+               })
+
+      assert updated_pool.id == pool.id
+      assert [] = all_enqueued(worker: CatalogSyncWorker)
+    end
+
+    test "final inactive pool assignment edit enqueues no catalog sync" do
+      %{user: owner} = bootstrap_owner_fixture(%{"email" => "catalog-inactive-owner@example.com"})
+      scope = Scope.for_user(owner, ["instance_owner"])
+      pool = pool_fixture(%{slug: "catalog-inactive-sync", name: "Catalog Inactive Sync"})
+      %{assignment: assignment} = upstream_assignment_fixture(pool)
+
+      assert {:ok, updated_pool} =
+               PoolWorkflow.update_pool_with_related_settings(scope, pool, %{
+                 "name" => "Catalog Inactive Sync Updated",
+                 "status" => "disabled",
+                 "routing_strategy" => "bridge_ring",
+                 "upstream_assignment_ids" => [assignment.id],
+                 "api_key_ids" => []
+               })
+
+      assert updated_pool.status == "disabled"
+      assert [] = all_enqueued(worker: CatalogSyncWorker)
+    end
+
+    test "drained assignment edit sync refreshes model source metadata for new upstream" do
+      %{user: owner} =
+        bootstrap_owner_fixture(%{"email" => "catalog-freshness-owner@example.com"})
+
+      scope = Scope.for_user(owner, ["instance_owner"])
+      pool = pool_fixture(%{slug: "catalog-freshness-sync", name: "Catalog Freshness Sync"})
+
+      first_upstream =
+        start_upstream(FakeUpstream.json_response(%{"data" => [%{"id" => "gpt-immediate-sync"}]}))
+
+      second_upstream =
+        start_upstream(FakeUpstream.json_response(%{"data" => [%{"id" => "gpt-immediate-sync"}]}))
+
+      first =
+        active_upstream_assignment_fixture(pool, %{
+          chatgpt_account_id: "acct_immediate_first",
+          metadata: %{"base_url" => FakeUpstream.url(first_upstream)}
+        })
+
+      assert {:ok, %{models: [initial_model]}} = Catalog.sync_pool_catalog(pool)
+      assert initial_model.metadata["source_assignment_ids"] == [first.assignment.id]
+
+      second_identity =
+        active_upstream_identity_fixture(%{
+          chatgpt_account_id: "acct_immediate_second",
+          metadata: %{"base_url" => FakeUpstream.url(second_upstream)}
+        })
+
+      assert {:ok, _secret} =
+               Upstreams.store_encrypted_secret(second_identity, %{
+                 secret_kind: "access_token",
+                 plaintext: "catalog-test-token-second"
+               })
+
+      assert {:ok, _updated_pool} =
+               PoolWorkflow.update_pool_with_related_settings(scope, pool, %{
+                 "name" => "Catalog Freshness Sync Updated",
+                 "status" => "active",
+                 "routing_strategy" => "bridge_ring",
+                 "upstream_identity_ids" => [first.identity.id, second_identity.id],
+                 "api_key_ids" => []
+               })
+
+      assignments_by_identity =
+        pool
+        |> Upstreams.list_pool_assignments()
+        |> Map.new(&{&1.upstream_identity_id, &1})
+
+      second_assignment = Map.fetch!(assignments_by_identity, second_identity.id)
+      assert second_assignment.status == "active"
+      assert [job] = all_enqueued(worker: CatalogSyncWorker)
+      assert job.args == %{"pool_id" => pool.id, "trigger_kind" => "manual"}
+      assert %{success: 1, failure: 0} = Oban.drain_queue(queue: :jobs)
+
+      refreshed_model = Catalog.get_model_by_exposed_id(pool, "gpt-immediate-sync")
+      expected_assignment_ids = Enum.sort([first.assignment.id, second_assignment.id])
+
+      assert refreshed_model.source_assignment_count == 2
+      assert refreshed_model.metadata["source_assignment_ids"] == expected_assignment_ids
+    end
+  end
+
   describe "pool routing settings workflow" do
     test "creation persists request compression routing setting when enabled" do
       %{user: owner} = bootstrap_owner_fixture(%{"email" => "owner@example.com"})
@@ -157,4 +324,10 @@ defmodule CodexPooler.Admin.PoolWorkflowTest do
     |> Task.async()
     |> Task.await(5_000)
   end
+
+  defp start_upstream(mode) do
+    {:ok, upstream} = FakeUpstream.start_link(mode)
+    on_exit(fn -> FakeUpstream.stop(upstream) end)
+    upstream
+  end
 end