Defer dataclip search_vector indexing off the insert path (#4800)

The dataclips AFTER INSERT trigger built search_vector with jsonb_to_tsvector
on the synchronous insert path. For large bodies under load this could hold the
connection past the timeout and roll back the insert, so the dataclip was never
saved and the run's following events cascade-failed - losing the run.

Move vector building off the insert path, mirroring the log_lines approach:

- safe_jsonb_to_tsvector(regconfig, jsonb): COALESCE(body,'{}') so a NULL/wiped
  body yields ''::tsvector (never NULL, so it can't stick in the pending index),
  catching program_limit_exceeded -> ''::tsvector.
- Partial index dataclips_pending_search_idx over (inserted_at) WHERE
  search_vector IS NULL, built CONCURRENTLY (dataclips is unpartitioned).
- Drop the set_search_vector trigger and update_dataclip_search_vector function
  (down restores the program_limit_exceeded-catching version).
- DataclipSearchVectorWorker on a dedicated dataclip_search_indexing queue
  (concurrency 1) drains pending rows newest-first with FOR UPDATE SKIP LOCKED,
  snowballing when its per-run budget is exhausted, otherwise minute-ly cron.
  Uses english_nostop to match the read side (Lightning.Invocation).

Dataclip search is now eventually consistent. The insert no longer blocks on
or rolls back from vector building.

Author: stuartc

CHANGELOG.md

@@ -21,6 +21,13 @@ and this project adheres to
 
 ### Fixed
 
+- Dataclip inserts no longer roll back when building the full-text search vector
+  is slow. The `jsonb_to_tsvector` work that ran in an `AFTER INSERT` trigger
+  could hold the connection past the timeout and roll back the insert, losing
+  the whole run. The search vector is now built off the insert path by a
+  background `Lightning.Invocation.DataclipSearchVectorWorker` (dedicated
+  `dataclip_search_indexing` queue), making dataclip search eventually
+  consistent. [#4800](https://github.com/OpenFn/lightning/issues/4800)
 - 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/config/bootstrap.ex

@@ -270,7 +270,8 @@ defmodule Lightning.Config.Bootstrap do
        args: %{"type" => "monthly_project_digest"}},
       #  TODO - move this into an ENV?
       {"17 */2 * * *", Lightning.Projects, args: %{"type" => "data_retention"}},
-      {"*/10 * * * *", Lightning.KafkaTriggers.DuplicateTrackingCleanupWorker}
+      {"*/10 * * * *", Lightning.KafkaTriggers.DuplicateTrackingCleanupWorker},
+      {"* * * * *", Lightning.Invocation.DataclipSearchVectorWorker}
     ]
 
     cleanup_cron =
@@ -300,7 +301,8 @@ defmodule Lightning.Config.Bootstrap do
         workflow_failures: 1,
         background: 1,
         history_exports: 1,
-        ai_assistant: 10
+        ai_assistant: 10,
+        dataclip_search_indexing: 1
       ]
 
     # https://plausible.io/ is an open-source, privacy-friendly alternative to

lib/lightning/invocation/dataclip_search_vector_worker.ex

@@ -0,0 +1,96 @@
+defmodule Lightning.Invocation.DataclipSearchVectorWorker do
+  @moduledoc """
+  Backfills the full-text `search_vector` on `dataclips` rows.
+
+  Dataclips are inserted with `search_vector` left `NULL`; the vector is built
+  here rather than on the insert path. Building it inline was risky:
+  `jsonb_to_tsvector` over a large dataclip body is slow and runs inside the
+  transaction that persists the run, so a slow (or failing) vector build could
+  roll back the dataclip insert and lose the run (#4800). Deferring it keeps
+  `jsonb_to_tsvector` off that hot path. Search is eventually consistent as a
+  result, typically catching up within a minute.
+
+  Two database objects support this: `safe_jsonb_to_tsvector(regconfig, jsonb)`,
+  which builds the vector from the dataclip body while tolerating NULL and
+  oversized input, and a partial index over `search_vector IS NULL`, which keeps
+  locating pending rows cheap as the table grows. Vectors use the
+  `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
+  and enqueues an immediate follow-up ("snowball"); otherwise the minute-ly cron
+  tick keeps pace. The worker runs on the dedicated `dataclip_search_indexing`
+  queue at concurrency 1, so only one job executes at a time, and the cron tick
+  and the snowball carry distinct `trigger` args, so job uniqueness allows one of
+  each to queue but never a duplicate. The queue is separate from the log_lines
+  `search_indexing` queue so dataclip indexing can't be starved by a log-line
+  backlog.
+  """
+
+  use Oban.Worker,
+    queue: :dataclip_search_indexing,
+    priority: 1,
+    max_attempts: 10,
+    # Restrict uniqueness to queued states. Oban's defaults also dedup against
+    # :executing/:completed, so a running snowball would match itself and fail
+    # to enqueue its successor — breaking the chain after one hop.
+    unique: [period: 55, keys: [:trigger], states: [:available, :scheduled]]
+
+  alias Lightning.Repo
+
+  require Logger
+
+  @batch_size 2_500
+  # Per-run budget.
+  @max_batches 10
+
+  @drain_sql """
+  WITH pending AS (
+    SELECT id FROM dataclips
+    WHERE search_vector IS NULL
+    ORDER BY inserted_at DESC
+    LIMIT $1 FOR UPDATE SKIP LOCKED
+  )
+  UPDATE dataclips d
+  SET search_vector =
+    safe_jsonb_to_tsvector('public.english_nostop'::regconfig, d.body)
+  FROM pending p WHERE d.id = p.id
+  """
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{}) do
+    {filled, budget_exhausted?} = drain(0, 0)
+
+    Logger.info(fn ->
+      # coveralls-ignore-start
+      "Invocation.DataclipSearchVectorWorker filled #{filled} search_vector row(s)."
+      # coveralls-ignore-stop
+    end)
+
+    if budget_exhausted? do
+      # Budget exhausted, so backlog likely remains: enqueue an immediate
+      # follow-up rather than waiting for the next cron tick.
+      Oban.insert(Lightning.Oban, __MODULE__.new(%{"trigger" => "snowball"}))
+    end
+
+    {:ok, filled}
+  end
+
+  # 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}
+    else
+      drain(filled + num_rows, batches + 1)
+    end
+  end
+end

priv/repo/migrations/20260530184652_add_safe_jsonb_to_tsvector_function.exs

@@ -0,0 +1,24 @@
+defmodule Lightning.Repo.Migrations.AddSafeJsonbToTsvectorFunction do
+  use Ecto.Migration
+
+  def up do
+    # Deliberately not STRICT: a STRICT function returns NULL for a NULL doc,
+    # which would leave search_vector NULL forever and stuck in the pending
+    # index (e.g. a wiped dataclip with a NULL body). COALESCE instead so the
+    # result is always a non-NULL tsvector.
+    execute("""
+    CREATE OR REPLACE FUNCTION safe_jsonb_to_tsvector(config regconfig, doc jsonb)
+    RETURNS tsvector LANGUAGE plpgsql IMMUTABLE AS $$
+    BEGIN
+      RETURN jsonb_to_tsvector(config, COALESCE(doc, '{}'::jsonb), '"all"');
+    EXCEPTION WHEN program_limit_exceeded THEN
+      RETURN ''::tsvector;
+    END;
+    $$;
+    """)
+  end
+
+  def down do
+    execute("DROP FUNCTION IF EXISTS safe_jsonb_to_tsvector(regconfig, jsonb);")
+  end
+end

priv/repo/migrations/20260530184653_add_dataclips_pending_search_index.exs

@@ -0,0 +1,23 @@
+defmodule Lightning.Repo.Migrations.AddDataclipsPendingSearchIndex do
+  use Ecto.Migration
+
+  @disable_ddl_transaction true
+  @disable_migration_lock true
+
+  def up do
+    # Partial index that keeps "find pending rows" cheap for the background
+    # indexing worker. dataclips is an unpartitioned table, so this is a single
+    # CONCURRENTLY-built index (no per-partition build-and-attach dance). The
+    # index stays small because existing rows already have a non-NULL
+    # search_vector, so only freshly-inserted, not-yet-indexed rows match.
+    execute("""
+    CREATE INDEX CONCURRENTLY IF NOT EXISTS dataclips_pending_search_idx
+    ON dataclips (inserted_at)
+    WHERE search_vector IS NULL
+    """)
+  end
+
+  def down do
+    execute("DROP INDEX CONCURRENTLY IF EXISTS dataclips_pending_search_idx")
+  end
+end

priv/repo/migrations/20260530184654_drop_dataclip_search_vector_trigger.exs

@@ -0,0 +1,40 @@
+defmodule Lightning.Repo.Migrations.DropDataclipSearchVectorTrigger do
+  use Ecto.Migration
+
+  def up do
+    execute("SET lock_timeout = '5s'")
+    execute("DROP TRIGGER IF EXISTS set_search_vector ON dataclips")
+    execute("DROP FUNCTION IF EXISTS update_dataclip_search_vector()")
+  end
+
+  def down do
+    # Restore the program_limit_exceeded-catching version of the function
+    # (from 20250219122902), not the original naive one.
+    execute("""
+    CREATE OR REPLACE FUNCTION update_dataclip_search_vector()
+    RETURNS trigger
+    LANGUAGE plpgsql
+    AS $function$
+    BEGIN
+      BEGIN
+        UPDATE dataclips
+        SET search_vector = jsonb_to_tsvector('english_nostop', body, '"all"')
+        WHERE id = NEW.id;
+      EXCEPTION
+        WHEN program_limit_exceeded THEN
+          RAISE NOTICE 'Message too long for tsvector at id: %. Error: %', NEW.id, SQLERRM;
+      END;
+
+      RETURN NEW;
+    END;
+    $function$;
+    """)
+
+    execute("""
+    CREATE TRIGGER set_search_vector
+    AFTER INSERT ON dataclips FOR EACH ROW
+    WHEN (NEW."search_vector" IS NULL)
+    EXECUTE PROCEDURE update_dataclip_search_vector();
+    """)
+  end
+end