Port size limit alert from v1

Dropped-output-state notification (port from OpenFn V1)

Context

In OpenFn V1, project users with failure notifications enabled received an
email whenever a step's output state was dropped because it exceeded the
platform storage limit — so they knew the data wasn't persisted and would not
flow to downstream steps or be available for replay.

This feature does not exist in Lightning today. The worker
(@openfn/ws-worker) already detects oversize step outputs and signals them on
step:complete with an extra field — output_dataclip_error: "DATACLIP_TOO_LARGE" — paired with output_dataclip: null (no body sent) so
the message stays under the WebSocket frame limit. Lightning currently
ignores this field: it isn't cast in the CompleteStep embedded schema, it
isn't persisted on the step, and no alert is dispatched.

The intended outcome: when the worker reports DATACLIP_TOO_LARGE for a step,
Lightning records it on the step, the run continues, and at most one email per
run is sent — coalesced across all steps that dropped — to project users who
already have failure_alert enabled.

We must not reuse the dataclip wiped_at state for this. wiped_at
indicates a project-level data-privacy/retention drop (save_dataclips: false)
and conflating it with a size drop would muddle that signal.

---

Decisions (per user)

• Detection driven by the worker's existing output_dataclip_error: "DATACLIP_TOO_LARGE" signal on step:complete. No worker protocol changes
  in this PR.
• Persist output_dataclip_error on the step (new column) and
  final_dataclip_error on the run (new column), parallel structure. No
  dataclip row is created for the dropped output — current
  maybe_save_dataclip/2 already handles output_dataclip: nil by returning
  {:ok, nil}; CompleteRun.save_final_dataclip/3 is similarly skipped when
  final_state is absent.
• Do not touch wiped_at.
• Drop email only. No warning/proximity email.
• Reuse ProjectUser.failure_alert as the opt-in. No new field, no UI
  change.
• One email per run. Coalesce drops across all steps into a single email
  per recipient at run completion.
• The worker's message stays under the frame limit (it sent null instead of
  the oversize body), so no change to max_frame_size is needed.

Flow

worker                                Lightning
  │
  │ step:complete {
  │   output_dataclip: null,
  │   output_dataclip_error:
  │     "DATACLIP_TOO_LARGE", ...
  │ }
  ├──────────────────────► RunChannel.handle_in("step:complete")
  │                              │
  │                              ▼
  │                          Handlers.CompleteStep
  │                          • cast :output_dataclip_error
  │                          • maybe_save_dataclip → no-op (body nil)
  │                          • Step.finished sets
  │                              output_dataclip_error="DATACLIP_TOO_LARGE",
  │                              output_dataclip_id=nil
  │
  │ run:complete {
  │   state: "ok",
  │   final_state: null,
  │   final_dataclip_error:
  │     "DATACLIP_TOO_LARGE", ...
  │ }
  ├──────────────────────► RunChannel.handle_in("run:complete")
                                 │
                                 ▼
                             Handlers.CompleteRun
                             • cast :final_dataclip_error
                             • resolve_final_dataclip → no insert when
                               final_state is nil
                             • Run.complete sets
                                 final_dataclip_error="DATACLIP_TOO_LARGE",
                                 final_dataclip_id=nil
                                 │
                                 ▼
                             FailureAlerter.alert_on_failure(run)        (existing)
                             DataclipDropAlerter.alert_on_run_complete(run) (new)
                                 │
                                 ▼ (1 query + a field check)
                             dropped? = run.final_dataclip_error != nil
                                       OR EXISTS step WHERE run_id=? AND
                                          output_dataclip_error IS NOT NULL
                                 │
                                 ▼ if dropped?:
                             ProjectLimiter.limit_failure_alert  (entitlement)
                             get_users_to_alert_for_project       (failure_alert opt-in)
                             per recipient → Hammer (separate bucket)
                                          → Lightning.DataclipDropEmail.deliver

The drop alerter and the failure alerter are independent: a run that both
fails and drops sends two emails (each conveys distinct information). The
drop alerter still fires when state == :success because a successful run
can still drop oversize outputs.

Implementation

1. Schema + migration

Add nullable string columns: output_dataclip_error on steps and
final_dataclip_error on runs. Index not needed (the alerter scans for one
run at a time and reads the run row directly).

• priv/repo/migrations/<timestamp>_add_dataclip_error_columns.exs:

  alter table(:steps) do
    add :output_dataclip_error, :string
  end
  alter table(:runs) do
    add :final_dataclip_error, :string
  end
  

• lib/lightning/invocation/step.ex
  • Add field :output_dataclip_error, :string (next to error_type, line 53).
  • Add to @derive Jason fields list.
  • Add to Step.finished/2 cast list and to Step.changeset/2 cast list.
• lib/lightning/runs/run.ex
  • Add field :final_dataclip_error, :string (next to error_type, line 111).
  • Add to Run.complete/2 cast list (line 190).
  • Update @derive Jason fields if applicable.

2. Wire the worker payload through

lib/lightning/runs/handlers.ex:

CompleteStep module:

• Add field :output_dataclip_error, :string to the embedded schema (around
  line 350, sibling to error_type).
• Add :output_dataclip_error to the cast/3 list in new/2 (around line
  357).
• In to_step_params/1 (line 425), include :output_dataclip_error in the
  Map.take/2 so it's passed to Step.finished/2.
• maybe_save_dataclip/2 needs no change: the existing
  output_dataclip: nil head (line 458–463) already returns {:ok, nil},
  which is the correct behavior when the worker dropped the body.

CompleteRun module:

• Add field :final_dataclip_error, :string to the embedded schema (around
  line 82, sibling to error_type).
• Add :final_dataclip_error to the cast/3 list in new/1 (around line
  110).
• In to_run_params/1 (line 142), include :final_dataclip_error in the
  Map.take/2 so it's passed to Run.complete/2.
• resolve_final_dataclip/2 needs no change: when final_state is nil
  the third clause (line 181) already returns {:ok, to_run_params(...)}
  without inserting a dataclip — correct behavior when the worker dropped.

3. New module: Lightning.DataclipDropAlerter

lib/lightning/dataclip_drop_alerter.ex — modeled on
lib/lightning/pipeline/failure_alerter.ex.

defmodule Lightning.DataclipDropAlerter do
  @moduledoc false
  use LightningWeb, :verified_routes

  import Ecto.Query

  alias Lightning.Invocation.Step
  alias Lightning.Projects.ProjectLimiter
  alias Lightning.Repo
  alias Lightning.Run

  def alert_on_run_complete(nil), do: nil

  def alert_on_run_complete(%Run{} = run) do
    drops = dropped_steps(run)
    final_drop = run.final_dataclip_error

    if drops == [] and is_nil(final_drop) do
      nil
    else
      dispatch(run, drops, final_drop)
    end
  end

  defp dropped_steps(%Run{id: run_id}) do
    from(s in Step,
      join: rs in Lightning.RunStep, on: rs.step_id == s.id,
      join: j in assoc(s, :job),
      where: rs.run_id == ^run_id,
      where: not is_nil(s.output_dataclip_error),
      select: %{step_id: s.id, job_name: j.name,
                error: s.output_dataclip_error}
    )
    |> Repo.all()
  end

  defp dispatch(run, drops, final_drop) do
    workflow = run.work_order.workflow

    if :ok == ProjectLimiter.limit_failure_alert(workflow.project_id) do
      project = Lightning.Projects.get_project!(workflow.project_id)

      Lightning.Accounts.get_users_to_alert_for_project(%{id: workflow.project_id})
      |> Enum.each(&alert(&1, run, workflow, project, drops, final_drop))
    end
  end

  defp alert(recipient, run, workflow, project, drops, final_drop) do
    [time_scale: time_scale, rate_limit: rate_limit] =
      Application.fetch_env!(:lightning, __MODULE__)

    bucket_key = "drop::#{workflow.id}::#{recipient.id}"

    case Hammer.check_rate(bucket_key, time_scale, rate_limit) do
      {:allow, count} ->
        run_url = url(LightningWeb.Endpoint,
          ~p"/projects/#{workflow.project_id}/runs/#{run.id}")

        deliver_or_decrement(bucket_key, time_scale, rate_limit, %{
          recipient: recipient,
          run_id: run.id,
          run_url: run_url,
          workflow_name: workflow.name,
          workflow_id: workflow.id,
          project_name: project.name,
          dropped_steps: drops,
          final_state_dropped: not is_nil(final_drop),
          count: count,
          time_scale: time_scale,
          rate_limit: rate_limit
        })

      {:deny, _} ->
        nil
    end
  end

  defp deliver_or_decrement(bucket_key, time_scale, rate_limit, body_data) do
    case Lightning.DataclipDropEmail.deliver(body_data.recipient.email, body_data) do
      {:ok, _meta} -> nil
      _ ->
        Hammer.check_rate_inc(bucket_key, time_scale, rate_limit, -1)
        nil
    end
  end
end

Notes:

• bucket_key is prefixed with "drop::" so the rate counter is disjoint
  from FailureAlerter's bucket — a workflow that's both failing and dropping
  doesn't silence one feed with the other.
• Reuses Lightning.Accounts.get_users_to_alert_for_project/1 (which already
  filters on ProjectUser.failure_alert == true).
• Reuses ProjectLimiter.limit_failure_alert/1 for the entitlement gate
  (Action.type :alert_failure). Same toggle, same entitlement. If we want
  separate entitlement later, that's a one-line follow-up adding
  :alert_dataclip_drop to Lightning.Extensions.UsageLimiting.Action.

4. New email module + template + view

• lib/lightning/dataclip_drop_email.ex — mirrors
  lib/lightning/pipeline/failure_email.ex. deliver(email, body_data),
  subject e.g. "\"<workflow>\" (<project>) dropped output state".
• lib/lightning_web/views/dataclip_drop_notifier_view.ex —
  defmodule LightningWeb.DataclipDropNotifierView do use LightningWeb, :view end.
• lib/lightning_web/templates/dataclip_drop_notifier/drop.html.heex —
  body covering: hi @recipient.first_name, in @workflow_name (@project_name)
  one or more outputs were too large to store, list of @dropped_steps
  (job names), explicit "the output state was not persisted and will not be
  available to downstream steps or for replay", link to @run_url,
  the standard @count / @duration / @rate_limit rate-limit footer.

5. Trigger from the channel

lib/lightning_web/channels/run_channel.ex, in handle_in("run:complete", …)
at line 121, immediately after Lightning.FailureAlerter.alert_on_failure/1:

run_with_preloads
|> Lightning.FailureAlerter.alert_on_failure()

run_with_preloads
|> Lightning.DataclipDropAlerter.alert_on_run_complete()

The existing preload (line 116–118) loads :work_order, [:workflow, :trigger]
and :log_lines. dropped_steps/1 runs its own query, so no extra preload
needed.

6. Configuration

config/config.exs — sibling of the FailureAlerter block at line 175:

config :lightning, Lightning.DataclipDropAlerter,
  time_scale: 5 * 60_000,
  rate_limit: 3

(Same defaults as failure alerts; can be tuned per environment in
config/runtime.exs later if desired.)

No endpoint.ex max_frame_size change. No bootstrap.ex env-var change.
The existing MAX_DATACLIP_SIZE_MB env continues to govern the limit (the
worker reads its own copy of that env).

Files to modify / add

Modify:

• lib/lightning/invocation/step.ex — add output_dataclip_error field, derive
  it, cast in finished/2 and changeset/2.
• lib/lightning/runs/run.ex — add final_dataclip_error field, cast in
  Run.complete/2.
• lib/lightning/runs/handlers.ex — extend CompleteStep and CompleteRun
  embedded schemas + cast/3 + to_*_params/1 to thread the new fields
  through.
• lib/lightning_web/channels/run_channel.ex — call alerter near line 121.
• config/config.exs — Hammer config block for the new alerter near line 175.

Add:

• priv/repo/migrations/<timestamp>_add_dataclip_error_columns.exs
• lib/lightning/dataclip_drop_alerter.ex
• lib/lightning/dataclip_drop_email.ex
• lib/lightning_web/views/dataclip_drop_notifier_view.ex
• lib/lightning_web/templates/dataclip_drop_notifier/drop.html.heex

No changes to:

• lib/lightning_web/endpoint.ex (frame size unchanged)
• lib/lightning/config.ex, lib/lightning/config/bootstrap.ex (no new env)
• ProjectUser schema, project-members UI
• CompleteRun handler, dataclip wiped_at semantics
• Lightning.Extensions.UsageLimiting.Action (reusing :alert_failure)

Tests

• test/lightning/runs/handlers_test.exs:
  • CompleteStep: worker sends output_dataclip: null +
    output_dataclip_error: "DATACLIP_TOO_LARGE" → step persisted with the
    error string, no dataclip row created, exit_reason still set from worker.
  • CompleteStep: worker sends normal output → no output_dataclip_error
    set (regression).
  • CompleteRun: worker sends final_state: null + final_dataclip_error: "DATACLIP_TOO_LARGE" → run persisted with the error string,
    final_dataclip_id is nil, no dataclip row created.
  • CompleteRun: worker sends final_state: %{...} → no
    final_dataclip_error set (regression).
• test/lightning/dataclip_drop_alerter_test.exs:
  • No drops on the run (no step drops, no final drop) → no email.
  • Only step drops → one email per recipient, no final-state mention.
  • Only final-state drop → one email per recipient mentioning the run output.
  • Both step + final drops on one run → exactly one email per recipient
    that lists both.
  • Recipient with failure_alert: false → not emailed.
  • Hammer rate limit allows N emails then denies (mirrors
    failure_alerter_test.exs).
  • Hammer bucket disjoint from FailureAlerter (drop hits don't decrement
    the failure bucket and vice-versa).
  • Email-delivery error decrements the counter (use the existing test
    pattern that asserts on Hammer.check_rate/3).
• test/lightning/dataclip_drop_email_test.exs — template rendering
  (recipient salutation, workflow + project, list of step names, run link).
• test/lightning_web/channels/run_channel_test.exs — assert the alerter is
  invoked on run:complete (mock or use Mox; mirror how
  FailureAlerter.alert_on_failure is verified today, if it is).

Look at test/lightning/failure_alerter_test.exs and
test/lightning/failure_email_test.exs (if present) for the closest
patterns.

Verification (manual end-to-end)

1. With a worker version that already sends output_dataclip_error: "DATACLIP_TOO_LARGE", set MAX_DATACLIP_SIZE_MB=1 in .env (and the
   matching env on the worker side) and restart both.
2. As a project admin (with failure_alert enabled in project settings), run
   a workflow whose first job emits ~1.5 MB of state, with a downstream job.
3. Expect: one drop email; verify the dropping step row has
   output_dataclip_error = "DATACLIP_TOO_LARGE" and output_dataclip_id IS NULL; verify the downstream step sees nil input and the run state is what
   the worker reported. Also run a workflow whose final job emits ~1.5 MB and
   verify the run row has final_dataclip_error = "DATACLIP_TOO_LARGE" and
   final_dataclip_id IS NULL.
4. Re-run the dropping workflow 5× in a minute → expect exactly 3 emails
   (rate limit), then silence.
5. Verify a project user who has not enabled failure_alert receives nothing.
6. Verify a workflow that fails AND drops sends both the failure email and
   the drop email (two distinct emails, two distinct rate buckets).
7. Verify mix verify passes (format, credo --strict --all, dialyzer).

Start by reading lib/lightning/pipeline/failure_alerter.ex and the CompleteStep and CompleteRun handlers in lib/lightning/runs/handlers.ex, then trace completion in lib/lightning_web/channels/run_channel.ex. Review the step and run schemas, migration, email modules and templates named in the issue. Done means size-drop errors are persisted, runs continue, and eligible users receive at most one coalesced drop email per run.