[#22] On-Chain Event Ingestion Pipeline with Transactional Outbox & Replay - #35
Conversation
|
@meshackyaro PR opened for review. |
Thank you for tackling this complex feature, @realvic22. The implementation demonstrates a solid understanding of the transactional outbox pattern and Spring Boot best practices. The core architecture is sound, with well-designed entities, repositories, and service logic. However, there are four critical blockers and a significant gap in test coverage that must be addressed before this PR can be merged. The good news: these are all fixable with focused effort. Once resolved, this PR will provide a robust foundation for on-chain event processing. Blocking Issues
The replay() method iterates over events, modifies their status, attempts, errors, and timestamps in memory, but never persists these changes back to the database. This means when a caller invokes the /replay endpoint, the operation will return a success count, but the actual event records will remain unchanged in the database. To fix this: After updating each event and its corresponding outbox record, you need to explicitly persist those changes. Either call save() on the repositories after each update, or refactor to use @Modifying bulk update queries for better performance at scale.
The repository queries use LockModeType.PESSIMISTIC_WRITE to ensure only one worker processes an event, but there is no test verifying this mechanism actually works. Without a concurrency test, you risk race conditions in production where multiple workers claim and process the same event simultaneously. To fix this: Add an integration test (using @SpringBootTest with a real test database like H2 or TestContainers) that spawns multiple concurrent threads attempting to claim the same PENDING event. Verify that only one thread successfully locks and retrieves the event, while the other thread(s) block or return empty results. This is an explicit requirement from issue #22's acceptance criteria: "Endpoints/services are covered by tests including failure and concurrency paths."
The ingest() method checks for an existing event by eventKey first, which is good. However, if any unexpected error occurs after that check but before the event is successfully saved (e.g., a database constraint violation, null pointer, or connection error), the exception will bubble up to the caller without re-checking for idempotency. This means a caller could retry after a transient error and either get a different result (error vs. success) or, in a race condition scenario, a duplicate event might slip in. To fix this: Wrap the event save and outbox creation in error handling that ensures idempotency is maintained. If a data integrity violation occurs (e.g., duplicate eventKey race), attempt to retrieve and return the existing event instead of throwing. Handle unexpected errors gracefully to maintain the promise of idempotent ingestion.
The PR currently includes only one unit test (ingest idempotency with mocked repositories), but issue #22's acceptance criteria explicitly require: Tests covering "endpoints/services... including failure and concurrency paths" Integration tests verifying transactional outbox semantics (that events and outbox records are saved atomically) A Quick Summary These must address before this PR can be merged:
Once these are addressed, the implementation will meet the acceptance criteria and be ready for code review and merge. |
… error handling, add comprehensive tests - Fix replay() to explicitly call saveAll() to persist event/outbox state changes - Improve ingest() error handling: catch DataIntegrityViolationException and perform fallback lookup to guarantee idempotency even during race conditions - Add ChainEventHandler functional interface for extensible and testable event processing pipeline - Add comprehensive integration tests (ChainEventServiceIntegrationTest): - Transactional outbox semantics (event + outbox created atomically) - Happy-path processing (PENDING -> PROCESSED, outbox COMPLETED) - Failure path: dead-letter transition after MAX_ATTEMPTS (5) retries - Exponential backoff on transient failures - Replay persistence and reprocessing verification - Concurrency test: pessimistic locking prevents duplicate processing - Already-processed events are not claimed again - Expand unit tests (ChainEventServiceTest) covering: - Idempotency when event already exists - DataIntegrityViolationException with fallback lookup - Unexpected exception propagation - Replay explicit saveAll verification
|
Thank you for the thorough review, @meshackyaro. Here's how each blocker has been addressed: 1. Replay Operation Does Not Persist ChangesFixed in 2. Missing Concurrency Test for Pessimistic LockingAdded in
An additional test ( 3. Ingest Error Handling Does Not Guarantee IdempotencyFixed in
These paths are covered by unit tests in 4. Insufficient Test CoverageAdded comprehensive integration tests (
Additional Changes
CI ValidationJava is unavailable in this environment so |
- Use REQUIRES_NEW inserter so Postgres unique-key races don't abort the outer TX - Explicitly save event/outbox on process + replay (no reliance on dirty-check alone) - Disable scheduling in integration tests; unique event keys; mock reset - Fix dead-letter test (seed attempts=4); non-flaky backoff assertion - Update unit tests for inserter-based constructor
… scan Also drop unused ObjectMapper from ChainEventService constructor.
CI statusCommit Later pushes are stuck on action_required (fork PR workflow approval). @meshackyaro could you approve the pending workflow run for the latest commit so CI can validate the hardened changes? What was hardened (to prevent further flakes)
|
meshackyaro
left a comment
meshackyaro
left a comment
There was a problem hiding this comment.
Excellent Work on the On-Chain Event Ingestion Pipeline!
@realvic22, this is a well-executed implementation of a critical infrastructure component. I'd like to highlight what stands out:
Architecture & Design
Transactional Outbox Pattern: The dual-entity approach (OnChainEvent + OutboxEvent) demonstrates strong understanding of distributed systems principles. This ensures event durability and enables reliable downstream processing — exactly what you want for blockchain-related operations.
Idempotency: The ChainEventInserter with REQUIRES_NEW propagation is a smart touch. By isolating the insert in a nested transaction, you allow the outer transaction to recover gracefully from race conditions on the unique constraint. The fallback retry logic is clean and correct.
Comprehensive Error Handling: The exponential backoff strategy (1 << Math.min(attempts, 6)) and dead-letter queue (5-attempt limit) show you've thought through production failure modes. This will save the team pain later.
Code Quality
- Well-structured repositories: Pessimistic locking with
@Lock(LockModeType.PESSIMISTIC_WRITE)and proper ordering ensures consistent event processing across concurrent workers. - Clean API contracts: Request/response DTOs with validation (
@NotBlank,@PositiveOrZero) are solid. - Thorough testing: 245 lines of integration tests covering idempotency, replay, concurrent access, and failure paths. The concurrent pessimistic-locking test is particularly impressive.
Strategic Touches
- Scheduled polling with configurable delay (
chain.events.poll-delay-ms) shows you designed for operational flexibility. - Security config updates to expose the new endpoints appropriately.
- Replay functionality for deterministic re-processing — essential for debugging chain events.
The summary, comprehensive test suite, and thoughtful architecture make this a standout contribution. Well done bringing issue #22 to completion!
2 participants
Summary:
Validation:
Closes #22
@meshackyaro