Harden log_lines search_vector deferral

Follow-up hardening on the search_vector deferral, from a review pass:

- Fix the worker's snowball chain. Oban's default unique states include
  :executing and :completed, so a running snowball job matched itself when
  enqueueing its successor and the insert was silently deduped, breaking the
  chain after one hop. Restrict uniqueness to [:available, :scheduled].
- Make the pending-search index migration re-runnable: a failed CREATE INDEX
  CONCURRENTLY leaves an INVALID index that IF NOT EXISTS would skip and ATTACH
  would leave the parent invalid forever. Drop any invalid leftover first.
- Make safe_to_tsvector NULL-defensive and re-runnable (CREATE OR REPLACE,
  drop STRICT, COALESCE the doc) so it never returns NULL and leaves a row
  stuck in the pending set. Also the template for the dataclip fix.
- Add snowball uniqueness regression tests.

Author: stuartc

CHANGELOG.md

@@ -21,6 +21,11 @@ and this project adheres to
 
 ### Fixed
 
+- Reduce `run:log` channel timeouts under heavy log volume by moving `log_lines`
+  search indexing off the insert path. The full-text search vector is now
+  backfilled by a background worker rather than computed synchronously on every
+  insert, so log search is eventually-consistent (typically within a minute).
+  [#4425](https://github.com/OpenFn/lightning/issues/4425)
 - Channel join crashes when multiple users open the same workflow concurrently
   [#4802](https://github.com/OpenFn/lightning/issues/4802)
 - Fix `purge_deleted` Oban job crashing when a soft-deleted project has

lib/lightning/log_lines/search_vector_worker.ex

@@ -39,7 +39,14 @@ defmodule Lightning.LogLines.SearchVectorWorker do
     queue: :search_indexing,
     priority: 1,
     max_attempts: 10,
-    unique: [period: 55, keys: [:trigger]]
+    # `states` is restricted to the queued states on purpose. Oban's default
+    # unique states include `:executing` and `:completed`, which would make a
+    # running snowball job match *itself* when it tries to enqueue its
+    # successor, silently dedup the insert, and break the chain after a single
+    # hop. Limiting uniqueness to `:available`/`:scheduled` still guarantees at
+    # most one queued snowball (and one queued cron heartbeat, via the distinct
+    # `:trigger` key) while letting the executing job enqueue the next link.
+    unique: [period: 55, keys: [:trigger], states: [:available, :scheduled]]
 
   alias Lightning.Repo
 

priv/repo/migrations/20260530091125_add_safe_to_tsvector_function.exs

@@ -2,11 +2,15 @@ defmodule Lightning.Repo.Migrations.AddSafeToTsvectorFunction do
   use Ecto.Migration
 
   def up do
+    # Not STRICT: a STRICT function returns NULL (without running) when `doc` is
+    # NULL, which would leave the row's search_vector NULL forever and stuck in
+    # the pending index. COALESCE the doc instead so the function always yields
+    # a non-NULL tsvector. CREATE OR REPLACE keeps the migration re-runnable.
     execute("""
-    CREATE FUNCTION safe_to_tsvector(config regconfig, doc text) RETURNS tsvector
-    LANGUAGE plpgsql IMMUTABLE STRICT AS $$
+    CREATE OR REPLACE FUNCTION safe_to_tsvector(config regconfig, doc text) RETURNS tsvector
+    LANGUAGE plpgsql IMMUTABLE AS $$
     BEGIN
-      RETURN to_tsvector(config, doc);
+      RETURN to_tsvector(config, COALESCE(doc, ''));
     EXCEPTION WHEN program_limit_exceeded THEN
       RETURN ''::tsvector;
     END;

priv/repo/migrations/20260530091126_add_log_lines_pending_search_index.exs

@@ -31,6 +31,25 @@ defmodule Lightning.Repo.Migrations.AddLogLinesPendingSearchIndex do
   end
 
   defp create_partition_index(_num_partitions, part_num) do
+    # A failed CREATE INDEX CONCURRENTLY leaves an INVALID index behind. The
+    # IF NOT EXISTS below would then skip rebuilding it, and the subsequent
+    # ATTACH would leave the parent permanently INVALID. Drop any invalid
+    # leftover first so a re-run rebuilds it cleanly. (A valid index is kept;
+    # only invalid ones are dropped.)
+    execute("""
+    DO $$
+    BEGIN
+      IF EXISTS (
+        SELECT 1 FROM pg_class c
+        JOIN pg_index i ON i.indexrelid = c.oid
+        WHERE c.relname = 'log_lines_#{part_num}_pending_search_idx'
+          AND NOT i.indisvalid
+      ) THEN
+        EXECUTE 'DROP INDEX log_lines_#{part_num}_pending_search_idx';
+      END IF;
+    END $$;
+    """)
+
     execute("""
     CREATE INDEX CONCURRENTLY IF NOT EXISTS log_lines_#{part_num}_pending_search_idx
     ON log_lines_#{part_num} (timestamp)

test/lightning/log_lines/search_vector_worker_test.exs

@@ -143,4 +143,53 @@ defmodule Lightning.LogLines.SearchVectorWorkerTest do
       end)
     end
   end
+
+  describe "snowball uniqueness" do
+    # Regression: Oban's default unique states include :executing and :completed,
+    # so a running snowball job (state :executing) matched *itself* when it tried
+    # to enqueue its successor — the insert was silently deduped and the chain
+    # died after one hop. The worker restricts uniqueness to the queued states so
+    # an executing job can always enqueue the next link.
+    test "an executing snowball does not block enqueuing its successor" do
+      Oban.Testing.with_testing_mode(:manual, fn ->
+        {:ok, running} =
+          Oban.insert(
+            Lightning.Oban,
+            SearchVectorWorker.new(%{"trigger" => "snowball"})
+          )
+
+        # Mimic Oban marking the job as executing while perform/1 runs.
+        from(j in Oban.Job, where: j.id == ^running.id)
+        |> Repo.update_all(set: [state: "executing"])
+
+        {:ok, successor} =
+          Oban.insert(
+            Lightning.Oban,
+            SearchVectorWorker.new(%{"trigger" => "snowball"})
+          )
+
+        refute successor.conflict?
+        refute successor.id == running.id
+      end)
+    end
+
+    test "two queued snowballs are deduped to one" do
+      Oban.Testing.with_testing_mode(:manual, fn ->
+        {:ok, first} =
+          Oban.insert(
+            Lightning.Oban,
+            SearchVectorWorker.new(%{"trigger" => "snowball"})
+          )
+
+        {:ok, second} =
+          Oban.insert(
+            Lightning.Oban,
+            SearchVectorWorker.new(%{"trigger" => "snowball"})
+          )
+
+        assert second.conflict?
+        assert second.id == first.id
+      end)
+    end
+  end
 end