Cover dataclip search_vector budget-exhaustion path

Mirror the log_lines fix: make DataclipSearchVectorWorker batch_size/max_batches
configurable through the Lightning.Config seam (defaults 250/10 in config.exs,
2/2 in test.exs), restructure drain/2 into drain/4, and add a test exercising
the recursive drain, budget guard, and snowball enqueue.

Author: stuartc

config/config.exs

@@ -189,6 +189,8 @@ config :lightning, :claim_work_mem, nil
 
 config :lightning, :log_lines_search_indexing, batch_size: 2_500, max_batches: 10
 
+config :lightning, :dataclip_search_indexing, batch_size: 250, max_batches: 10
+
 config :lightning, Lightning.Runtime.RuntimeManager, start: false
 
 config :lightning, LightningWeb.CollectionsController,

config/test.exs

@@ -137,6 +137,8 @@ config :lightning, :is_resettable_demo, true
 # the per-run budget guard, exercising the snowball follow-up path.
 config :lightning, :log_lines_search_indexing, batch_size: 2, max_batches: 2
 
+config :lightning, :dataclip_search_indexing, batch_size: 2, max_batches: 2
+
 config :lightning, :github_app,
   app_id: "111111",
   app_name: "test-github",

lib/lightning/config.ex

@@ -111,6 +111,20 @@ defmodule Lightning.Config do
       Application.get_env(:lightning, :log_lines_search_indexing, [])
     end
 
+    @impl true
+    def dataclip_search_indexing_batch_size do
+      dataclip_search_indexing_config() |> Keyword.fetch!(:batch_size)
+    end
+
+    @impl true
+    def dataclip_search_indexing_max_batches do
+      dataclip_search_indexing_config() |> Keyword.fetch!(:max_batches)
+    end
+
+    defp dataclip_search_indexing_config do
+      Application.get_env(:lightning, :dataclip_search_indexing, [])
+    end
+
     @impl true
     def default_ecto_database_timeout do
       Application.get_env(:lightning, Lightning.Repo) |> Keyword.get(:timeout)
@@ -499,6 +513,8 @@ defmodule Lightning.Config do
   @callback activity_cleanup_chunk_size() :: integer()
   @callback log_lines_search_indexing_batch_size() :: pos_integer()
   @callback log_lines_search_indexing_max_batches() :: pos_integer()
+  @callback dataclip_search_indexing_batch_size() :: pos_integer()
+  @callback dataclip_search_indexing_max_batches() :: pos_integer()
   @callback default_ecto_database_timeout() :: integer()
   @callback repo_connection_token_signer() :: Joken.Signer.t()
   @callback reset_password_token_validity_in_days() :: integer()
@@ -621,6 +637,14 @@ defmodule Lightning.Config do
     impl().log_lines_search_indexing_max_batches()
   end
 
+  def dataclip_search_indexing_batch_size do
+    impl().dataclip_search_indexing_batch_size()
+  end
+
+  def dataclip_search_indexing_max_batches do
+    impl().dataclip_search_indexing_max_batches()
+  end
+
   def default_ecto_database_timeout do
     impl().default_ecto_database_timeout()
   end

lib/lightning/invocation/dataclip_search_vector_worker.ex

@@ -17,8 +17,9 @@ defmodule Lightning.Invocation.DataclipSearchVectorWorker do
   `english_nostop` config to match the read side (`Lightning.Invocation`), which
   queries with `to_tsquery('english_nostop', ...)`.
 
-  Each run drains pending rows newest-first, in batches of `@batch_size` up to
-  `@max_batches` per run. A run that exhausts its budget leaves backlog behind
+  Each run drains pending rows newest-first, in batches up to a per-run budget
+  (batch size and max batches are configurable via `Lightning.Config`). A run
+  that exhausts its budget leaves backlog behind
   and enqueues an immediate follow-up ("snowball"); otherwise the minute-ly cron
   tick keeps pace. The worker shares the `search_indexing` queue with
   `Lightning.LogLines.SearchVectorWorker`; that queue runs at concurrency 2, so
@@ -40,10 +41,6 @@ defmodule Lightning.Invocation.DataclipSearchVectorWorker do
 
   require Logger
 
-  @batch_size 250
-  # Per-run budget.
-  @max_batches 10
-
   @drain_sql """
   WITH pending AS (
     SELECT id FROM dataclips
@@ -59,7 +56,10 @@ defmodule Lightning.Invocation.DataclipSearchVectorWorker do
 
   @impl Oban.Worker
   def perform(%Oban.Job{}) do
-    {filled, budget_exhausted?} = drain(0, 0)
+    batch_size = Lightning.Config.dataclip_search_indexing_batch_size()
+    max_batches = Lightning.Config.dataclip_search_indexing_max_batches()
+
+    {filled, budget_exhausted?} = drain(0, 0, batch_size, max_batches)
 
     Logger.info(fn ->
       # coveralls-ignore-start
@@ -76,20 +76,20 @@ defmodule Lightning.Invocation.DataclipSearchVectorWorker do
     {:ok, filled}
   end
 
-  # Drains up to @max_batches batches, accumulating the number of rows filled.
+  # Drains up to max_batches batches, accumulating the number of rows filled.
   # Returns {filled, budget_exhausted?}. Stops early when a batch fills fewer
-  # than @batch_size rows (backlog drained).
-  defp drain(filled, batches) when batches >= @max_batches do
-    {filled, true}
-  end
-
-  defp drain(filled, batches) do
-    %{num_rows: num_rows} = Repo.query!(@drain_sql, [@batch_size])
-
-    if num_rows < @batch_size do
-      {filled + num_rows, false}
+  # than batch_size rows (backlog drained).
+  defp drain(filled, batches, batch_size, max_batches) do
+    if batches >= max_batches do
+      {filled, true}
     else
-      drain(filled + num_rows, batches + 1)
+      %{num_rows: num_rows} = Repo.query!(@drain_sql, [batch_size])
+
+      if num_rows < batch_size do
+        {filled + num_rows, false}
+      else
+        drain(filled + num_rows, batches + 1, batch_size, max_batches)
+      end
     end
   end
 end

test/lightning/invocation/dataclip_search_vector_worker_test.exs

@@ -95,8 +95,11 @@ defmodule Lightning.Invocation.DataclipSearchVectorWorkerTest do
     end
 
     test "drains all pending dataclips across a batch" do
+      # Stay within the tiny test budget (batch_size: 2, max_batches: 2) so a
+      # single run drains everything: batch 1 fills 2, batch 2 fills 1 (< 2) and
+      # stops without tripping the budget.
       dataclips =
-        for n <- 1..5 do
+        for n <- 1..3 do
           insert(:dataclip, body: %{"n" => "draindataclip#{n}"})
         end
 
@@ -105,7 +108,7 @@ defmodule Lightning.Invocation.DataclipSearchVectorWorkerTest do
                  search_vector_state(dataclip.id, "draindataclip1")
       end
 
-      assert {:ok, 5} = perform_job(DataclipSearchVectorWorker, %{})
+      assert {:ok, 3} = perform_job(DataclipSearchVectorWorker, %{})
 
       for dataclip <- dataclips do
         assert %{null?: false} =
@@ -123,6 +126,22 @@ defmodule Lightning.Invocation.DataclipSearchVectorWorkerTest do
       end)
     end
 
+    test "snowballs an immediate follow-up when the per-run budget is exhausted" do
+      # config/test.exs sets batch_size: 2, max_batches: 2. With 5 pending rows
+      # the run fills two full batches (4 rows), reaches max_batches, and trips
+      # the budget guard, leaving backlog behind and enqueuing a snowball.
+      for n <- 1..5, do: insert(:dataclip, body: %{"n" => "overflow#{n}"})
+
+      Oban.Testing.with_testing_mode(:manual, fn ->
+        assert {:ok, 4} = perform_job(DataclipSearchVectorWorker, %{})
+
+        assert_enqueued(
+          worker: DataclipSearchVectorWorker,
+          args: %{"trigger" => "snowball"}
+        )
+      end)
+    end
+
     test "is idempotent: a second run with nothing pending fills 0 and snowballs nothing" do
       for n <- 1..3, do: insert(:dataclip, body: %{"n" => "again#{n}"})