fix: resolve race condition in compound trigger evaluation#7
Conversation
Fixes two race conditions in compound trigger evaluation: 1. **Never-firing race** (transactional): When two child triggers fire concurrently in separate transactions, each only sees its own uncommitted insert due to READ COMMITTED isolation. Neither sees enough firings to trigger the parent. Fix: Use PostgreSQL advisory locks to serialize concurrent evaluations for the same compound trigger. 2. **Double-firing race** (autocommit): When both transactions see all firings, both delete and both fire the parent. Fix: Use DELETE ... RETURNING to make clearing a claim operation. Only the worker that successfully deletes the expected firings proceeds; others bail out. Based on the fix in PrefectHQ/nebula#10716. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
📝 WalkthroughWalkthroughThis pull request adds advisory locking for composite triggers to prevent race conditions when multiple workers concurrently evaluate and fire the same trigger. The implementation acquires a database lock before processing child firings and detects when another worker has already claimed the firings. Changes
Sequence Diagram(s)sequenceDiagram
participant Worker_1
participant Worker_2
participant Database
participant Trigger_Eval
rect rgba(200, 150, 100, 0.5)
Note over Worker_1,Trigger_Eval: Composite Trigger Evaluation with Locking
end
Worker_1 ->> Database: acquire_composite_trigger_lock()
activate Database
Database ->> Worker_1: Lock acquired
deactivate Database
Worker_2 ->> Database: acquire_composite_trigger_lock()
activate Database
Note over Database: Blocked (Worker_1 holds lock)
deactivate Database
Worker_1 ->> Trigger_Eval: Check child firings
Worker_1 ->> Database: clear_child_firings()
activate Database
Database ->> Worker_1: Returns deleted_ids
deactivate Database
Worker_1 ->> Trigger_Eval: Compare deleted_ids vs firing_ids
alt All IDs deleted
Worker_1 ->> Trigger_Eval: Fire parent trigger
else Race condition detected
Worker_1 ->> Trigger_Eval: Log debug, exit early
end
Worker_1 ->> Database: Release lock
Worker_2 ->> Database: Lock now acquired
activate Database
Database ->> Worker_2: Lock acquired
deactivate Database
Worker_2 ->> Trigger_Eval: Check child firings (already claimed)
Worker_2 ->> Trigger_Eval: Exit early (no firings available)
Worker_2 ->> Database: Release lock
Estimated Code Review Effort🎯 4 (Complex) | ⏱️ ~50 minutes Poem
🚥 Pre-merge checks | ✅ 1 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (1 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
@coderabbitai review |
1 similar comment
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Fix all issues with AI agents
In `@src/prefect/server/events/models/composite_trigger_child_firing.py`:
- Around line 139-157: The docstring and returned column are mismatched: the
delete query currently
.returning(db.CompositeTriggerChildFiring.child_trigger_id) but the docstring
(and expected semantics) say it should return the deleted child_firing_ids;
update the .returning(...) call to return
db.CompositeTriggerChildFiring.child_firing_id instead (and keep the final
return as set(result.scalars().all())), or if returning child_trigger_id was
intentional, change the docstring to state that the function returns
child_trigger_id values; locate the delete block referencing
db.CompositeTriggerChildFiring, trigger, and firing_ids to make this change.
- Around line 42-44: The advisory lock currently computes lock_key using
Python's non-deterministic hash(str(trigger.id)), which varies between
processes; change this to use the deterministic UUID integer form instead:
derive a stable 64-bit key from trigger.id by using its integer representation
(trigger.id.int) reduced into the 63-bit range (e.g., modulo 2**63 or bitmask)
and pass that as lock_key to the existing await session.execute(sa.text("SELECT
pg_advisory_xact_lock(:key)"), {"key": lock_key}) call so all workers calculate
the same advisory lock value for the same trigger.id.
🧹 Nitpick comments (2)
src/prefect/server/events/models/composite_trigger_child_firing.py (1)
30-32: Consider logging when lock acquisition is skipped due to missing bind.If
session.get_bind()returnsNone, the function silently returns without acquiring the lock. This could mask configuration issues where the race condition protection is expected but not active. A debug log would aid troubleshooting.📝 Suggested debug log
bind = session.get_bind() if bind is None: + import logging + logging.getLogger(__name__).debug( + "Skipping advisory lock acquisition: session has no bind" + ) returnsrc/prefect/server/events/triggers.py (1)
71-73: Consider usingget_loggerfrom prefect.logging for consistency.The module imports
get_loggerfromprefect.logging(Line 27) but initializes the logger using standardlogging.getLogger. Prefect'sget_loggermay apply additional configuration. Consider using the already-importedget_loggerfor consistency with the rest of the codebase.♻️ Suggested change
-import logging - -logger = logging.getLogger(__name__) +logger = get_logger(__name__)
| lock_key = hash(str(trigger.id)) % (2**63) | ||
| await session.execute( | ||
| sa.text("SELECT pg_advisory_xact_lock(:key)"), {"key": lock_key} |
There was a problem hiding this comment.
Critical: Python's hash() is not deterministic across processes, breaking the advisory lock.
Python randomizes the hash seed by default (PYTHONHASHSEED), so different worker processes will compute different lock keys for the same trigger.id. This defeats the purpose of the advisory lock since concurrent workers won't contend for the same lock.
Use the UUID's deterministic integer representation instead:
🐛 Proposed fix using UUID's int representation
- lock_key = hash(str(trigger.id)) % (2**63)
+ # UUID.int gives a deterministic 128-bit integer; truncate to signed bigint range
+ lock_key = trigger.id.int % (2**63)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| lock_key = hash(str(trigger.id)) % (2**63) | |
| await session.execute( | |
| sa.text("SELECT pg_advisory_xact_lock(:key)"), {"key": lock_key} | |
| # UUID.int gives a deterministic 128-bit integer; truncate to signed bigint range | |
| lock_key = trigger.id.int % (2**63) | |
| await session.execute( | |
| sa.text("SELECT pg_advisory_xact_lock(:key)"), {"key": lock_key} |
🤖 Prompt for AI Agents
In `@src/prefect/server/events/models/composite_trigger_child_firing.py` around
lines 42 - 44, The advisory lock currently computes lock_key using Python's
non-deterministic hash(str(trigger.id)), which varies between processes; change
this to use the deterministic UUID integer form instead: derive a stable 64-bit
key from trigger.id by using its integer representation (trigger.id.int) reduced
into the 63-bit range (e.g., modulo 2**63 or bitmask) and pass that as lock_key
to the existing await session.execute(sa.text("SELECT
pg_advisory_xact_lock(:key)"), {"key": lock_key}) call so all workers calculate
the same advisory lock value for the same trigger.id.
| ) -> set[UUID]: | ||
| """ | ||
| Delete the specified child firings and return the IDs that were actually deleted. | ||
|
|
||
| Returns the set of child_firing_ids that were successfully deleted. Callers can | ||
| compare this to the expected firing_ids to detect races and avoid double-firing | ||
| composite triggers. | ||
| """ | ||
| result = await session.execute( | ||
| sa.delete(db.CompositeTriggerChildFiring) | ||
| .filter( | ||
| db.CompositeTriggerChildFiring.automation_id == trigger.automation.id, | ||
| db.CompositeTriggerChildFiring.parent_trigger_id == trigger.id, | ||
| db.CompositeTriggerChildFiring.child_firing_id.in_(firing_ids), | ||
| ) | ||
| .returning(db.CompositeTriggerChildFiring.child_trigger_id) | ||
| ) | ||
|
|
||
| return set(result.scalars().all()) |
There was a problem hiding this comment.
Return column mismatch: docstring says child_firing_ids but code returns child_trigger_id.
The docstring states the function "Returns the set of child_firing_ids that were successfully deleted", but the .returning() clause fetches child_trigger_id instead. While the count comparison in the caller may coincidentally work, this is semantically incorrect and could cause subtle bugs if the comparison logic changes.
🐛 Proposed fix to return the correct column
.returning(db.CompositeTriggerChildFiring.child_trigger_id)
+ .returning(db.CompositeTriggerChildFiring.child_firing_id)
- .returning(db.CompositeTriggerChildFiring.child_trigger_id)Or if child_trigger_id is intentional, update the docstring:
"""
Delete the specified child firings and return the IDs that were actually deleted.
- Returns the set of child_firing_ids that were successfully deleted. Callers can
- compare this to the expected firing_ids to detect races and avoid double-firing
+ Returns the set of child_trigger_ids that were successfully deleted. Callers can
+ compare this count to the expected firing count to detect races and avoid double-firing
composite triggers.
"""📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| ) -> set[UUID]: | |
| """ | |
| Delete the specified child firings and return the IDs that were actually deleted. | |
| Returns the set of child_firing_ids that were successfully deleted. Callers can | |
| compare this to the expected firing_ids to detect races and avoid double-firing | |
| composite triggers. | |
| """ | |
| result = await session.execute( | |
| sa.delete(db.CompositeTriggerChildFiring) | |
| .filter( | |
| db.CompositeTriggerChildFiring.automation_id == trigger.automation.id, | |
| db.CompositeTriggerChildFiring.parent_trigger_id == trigger.id, | |
| db.CompositeTriggerChildFiring.child_firing_id.in_(firing_ids), | |
| ) | |
| .returning(db.CompositeTriggerChildFiring.child_trigger_id) | |
| ) | |
| return set(result.scalars().all()) | |
| ) -> set[UUID]: | |
| """ | |
| Delete the specified child firings and return the IDs that were actually deleted. | |
| Returns the set of child_firing_ids that were successfully deleted. Callers can | |
| compare this to the expected firing_ids to detect races and avoid double-firing | |
| composite triggers. | |
| """ | |
| result = await session.execute( | |
| sa.delete(db.CompositeTriggerChildFiring) | |
| .filter( | |
| db.CompositeTriggerChildFiring.automation_id == trigger.automation.id, | |
| db.CompositeTriggerChildFiring.parent_trigger_id == trigger.id, | |
| db.CompositeTriggerChildFiring.child_firing_id.in_(firing_ids), | |
| ) | |
| .returning(db.CompositeTriggerChildFiring.child_firing_id) | |
| ) | |
| return set(result.scalars().all()) |
🤖 Prompt for AI Agents
In `@src/prefect/server/events/models/composite_trigger_child_firing.py` around
lines 139 - 157, The docstring and returned column are mismatched: the delete
query currently .returning(db.CompositeTriggerChildFiring.child_trigger_id) but
the docstring (and expected semantics) say it should return the deleted
child_firing_ids; update the .returning(...) call to return
db.CompositeTriggerChildFiring.child_firing_id instead (and keep the final
return as set(result.scalars().all())), or if returning child_trigger_id was
intentional, change the docstring to state that the function returns
child_trigger_id values; locate the delete block referencing
db.CompositeTriggerChildFiring, trigger, and firing_ids to make this change.
4 participants
Benchmark PR from agentic-review-benchmarks#7
Summary by CodeRabbit
Release Notes
Bug Fixes
Tests