fix(outbox): add lease-based claim/reclaim to prevent stranded events - #432
Merged
YaronZaki merged 4 commits intoAug 22, 2026
Merged
Conversation
- Add claimed_at column to OutboxEvent for lease tracking - Atomic claim via UPDATE ... WHERE status IN (pending,failed) - Reclaim stale processing events older than 5 minutes - Validate UUID format in process_position_opened_task - Clear claimed_at on failure for proper retry - Alembic migration for claimed_at column - Tests for failed publish, stale reclaim, UUID validation Closes Quantarq#417
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
|
|
||
| class TestUUIDValidation: | ||
| def test_valid_uuid(self): | ||
| assert _is_valid_uuid(str(uuid.uuid4())) is True |
| assert _is_valid_uuid(str(uuid.uuid4())) is True | ||
|
|
||
| def test_invalid_uuid(self): | ||
| assert _is_valid_uuid("not-a-uuid") is False |
| assert _is_valid_uuid("not-a-uuid") is False | ||
|
|
||
| def test_empty_string(self): | ||
| assert _is_valid_uuid("") is False |
|
|
||
| def test_truncated_uuid(self): | ||
| valid = str(uuid.uuid4()) | ||
| assert _is_valid_uuid(valid[:30]) is False |
| process_position_opened_task(task, event_id=str(event.id)) | ||
|
|
||
| updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() | ||
| assert updated.status == "failed" |
|
|
||
| updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() | ||
| assert updated.status == "failed" | ||
| assert updated.claimed_at is None |
| updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() | ||
| assert updated.status == "failed" | ||
| assert updated.claimed_at is None | ||
| assert updated.retry_count == 1 |
| import sentry_sdk | ||
| from datetime import datetime, timedelta | ||
| from celery import Celery | ||
| from sqlalchemy import and_, or_ |
| def test_already_claimed_event_is_not_double_dispatched( | ||
| self, mock_init_db, mock_delay, mock_session_local, db_session | ||
| ): | ||
| now = datetime.now() |
…ry invocation, fix mock assertions
| relay.process_pending_events() | ||
|
|
||
| updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() | ||
| assert updated.status == "processing" |
|
|
||
| updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() | ||
| assert updated.status == "processing" | ||
| assert updated.claimed_at is not None |
| relay.process_pending_events() | ||
|
|
||
| updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() | ||
| assert updated.status == "processing" |
|
|
||
| updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() | ||
| assert updated.status == "processing" | ||
| assert updated.claimed_at is not None |
| relay.process_pending_events() | ||
|
|
||
| updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() | ||
| assert updated.status == "processing" |
|
|
||
| updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() | ||
| assert updated.status == "processing" | ||
| assert updated.claimed_at is not None |
| relay.process_pending_events() | ||
|
|
||
| updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() | ||
| assert updated.status == "processing" |
| relay.process_pending_events() | ||
|
|
||
| updated = db_session.query(OutboxEvent).filter(OutboxEvent.id == event.id).one() | ||
| assert updated.status == "pending" |
- Convert event_id to uuid.UUID in process_position_opened_task to fix
SQLite UUID string comparison ('str' object has no attribute 'hex')
- Use fresh sessions in tests that query after relay/task closes the
original session (DetachedInstanceError)
- Track .update() calls in test_relay_worker_dispatches_task to properly
simulate synchronize_session='fetch' behavior
The relay's finally block closes the session, invalidating ORM objects for post-relay assertions. Stubbing close() as a no-op keeps the session usable while preserving cleanup in the fixture teardown.
6 tasks
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Add a lease-based claim/reclaim mechanism to the outbox relay so events stranded in "processing" state after a crash are automatically reclaimed and re-queued within 5 minutes.
Related Issue
Closes #417
Change Type
Testing Done
Screenshots (if UI changes)
None
Environment Variables
None
Checklist
make lintpasses (pylint on changed.pyfiles)make testpasses (pytest inquantara/web_app/tests/)Closes #417)