[shivaam]

Trigger.clean_unused() runs synchronously in every triggerer supervisor loop. It previously removed every eligible trigger in one transaction, so large cleanups could hold many row locks and delay the triggerer heartbeat and concurrent database work.

This change deletes at most one configured batch of unreferenced triggers per triggerer loop and database transaction. Remaining triggers are handled in later loops, allowing the triggerer to resume runner servicing and heartbeat between batches.

The cleanup:

• Uses a configurable batch size that defaults to 500 and rejects non-positive values.
• Uses SKIP LOCKED so concurrent triggerers can select different rows.
• Rechecks task, asset, and callback references in the DELETE before removing a trigger.
• Uses a portable select-ID/delete-ID query shape on PostgreSQL, MySQL, and SQLite.
• Retries a failed cleanup using a fresh database transaction.

Tests cover empty, below-limit, exact-limit, and larger-than-limit backlogs; draining five rows across three calls (2, 2, 1); protected task, asset, and callback references; invalid configuration; and fresh retry transactions.

Trade-off

Cleanup no longer drains the complete backlog in one call. Remaining rows wait for later triggerer loops. The batch size can be increased when necessary, with the trade-off that larger batches create larger transactions and may hold more database locks.

Follow-ups

• The existing TaskInstance trigger-reference update is still unbounded. Batching it requires a separate concurrency design.
• callback.trigger_id still needs a portable index and Alembic migration. That database change is intentionally kept separate from this focused cleanup change.
• Cleanup database errors still escape the triggerer supervisor loop after retries are exhausted. This behavior predates this PR; a separate change can look and investigate this further.

Prior work: #68269 proposed the original batching direction. This implementation is a current-main rewrite because that draft predates #68244 and committed inside a caller-injected session.

Validation:

• Focused PostgreSQL and MySQL cleanup suites: 13 passed on each backend.
• PostgreSQL and MySQL 1,000-row probes each deleted two committed batches of [500, 500].
• Real scheduler/triggerer PostgreSQL validation covered locked rows, eventual drain, and protected references.
• Full SQLite trigger-model suite: 57 passed.
• Regular and manual changed-file checks, formatting, linting, mypy, configuration validation, and security checks passed.
• Rebased onto upstream/main at d0b8476825; all five contribution commits remained exact range-diff matches.

closes: #68243

---

Was generative AI tooling used to co-author this PR?

• Yes — Codex (GPT-5)

Generated-by: Codex (GPT-5) following the guidelines

[RishabhBajaj97]

Cross-reference: #74650 is a minimal, targeted fix for #74614 (triggerer exiting with PendingRollbackError after a transient metadata-DB disconnect). It adds a session.rollback() on DBAPIError/StaleDataError inside the existing run_with_db_retries() block in Trigger.clean_unused, mirroring the pattern already used by retry_db_transaction in airflow/utils/retries.py.

The restructure in this PR — moving the retry around a with create_session(scoped=False) block — would incidentally close #74614 as a side effect, since each retry attempt gets a fresh session. Flagging the overlap in case this PR's reviewers prefer to subsume the fix here (making #74650 redundant on merge) rather than carrying two parallel fixes.

---

Drafted-by: Claude Code (Opus 4.7) (human reviewed before posting)

[kaxil]

Both points from the last round are in: the DELETE re-checks all three referrers, and each clean_unused() call now deletes at most one batch. LGTM.

Two non-blocking notes:

• The batch_size <= 0 check runs inside run_once after load_triggers() has already claimed triggers. A value of 0 reads like "no batching" (that is how [state_store] state_cleanup_batch_size defines it), and here it crash-loops the triggerer, leaving its claimed triggers orphaned until the health check threshold passes. Validating it in TriggererJobRunner.__init__ next to capacity would fail before any DB work.
• Batch the triggerer's Trigger.clean_unused() cleanup deletes #68243 describes the scheduler side as already fixed, but Batch the scheduler's unreferenced-trigger cleanup deletes #68241 was closed unmerged. SchedulerJobRunner._remove_unreferenced_triggers still deletes the whole backlog in one statement every parsing_cleanup_interval (60s by default). That deserves a follow-up, and the follow-up should decide whether it reuses this option or gets its own [scheduler] one.

[github-actions]

Backport successfully created: v3-4-test

Note: As of Merging PRs targeted for Airflow 3.X
the committer who merges the PR is responsible for backporting the PRs that are bug fixes (generally speaking) to the maintenance branches.

In matter of doubt please ask in #release-management Slack channel.

Status	Branch	Result
✅	v3-4-test