Emit all messages across sequence number gaps in messagesWithBatch#517
Conversation
…ache#516 * The query windowing introduced by apache#180 treated a windowed batch returning fewer messages than batchSize as the end of the journal: bounded streams (recovery, currentEventsByPersistenceId) completed early and silently dropped every message beyond a gap left by deleted messages, and live streams stalled forever polling an empty window. apache#195 fixed only the gap at the head of the journal. * Rework the unfoldAsync state from FlowControl into a QueryPlan state machine (QueryRemaining, QueryWindow, PollRemaining, Complete): a short windowed batch is no longer conclusive and falls back to one query over the full remaining range, which gaps cannot hide messages from. * Keep windowing as the dense fast path of apache#180: after a full batch shows the journal to be dense, queries stay bounded to [from, from + batchSize]. * Span the full remaining range on the first query, crossing a deleted journal head (snapshot cleanup) in a single round trip, subsuming the apache#195 special case. * Poll the full remaining range when tailing: a windowed poll can never reach messages appended beyond a trailing gap. * Specify the gap state machine in MessagesWithBatchTest (core, in-memory journal stub) and the database-coupled behavior in MessagesWithBatchDatabaseContractTest (H2 plus the five integration databases): hard and soft deletes, a prefix purge through the journal's delete, and serialization failures.
mucciolo
changed the title
|
@mucciolo thanks very much for the PR. Would you have any interest in adding the equivalent tests to https://github.com/apache/pekko-persistence-r2dbc or alternatively to the persistence-tck that is part of https://github.com/apache/pekko? It would be great to ensure consistent behaviour across different persistence implementations. Feel free to say no. |
|
the oracle integration tests fail due to integrity constraints for nulls - we might need to change the DDL for oracle but on a quick check for this It seems feasible that we have a code issue that affects oracle usage |
Oracle treats the empty string as NULL, so inserting the synthetic corrupt row with writer = "" violated the NOT NULL constraint on EVENT_JOURNAL.WRITER (ORA-01400) and failed only the Oracle integration tests. The production write path always carries a writer UUID, so this is a test-fixture-only fix.
These files contain no Akka-derived code, so they should carry the full ASF header (the headerLicense the build stamps on new files) rather than the shorter "derived from Akka" variant, per a maintainer review note.
|
If that's only about porting tests, I'm open to looking at Persistence TCK. No promises, but it's not a no either. Mind sharing more about what you're thinking? |
It's just about sharing the test scenarios. |
Issue required? |
There was a problem hiding this comment.
Pull request overview
This PR fixes a regression in BaseJournalDaoWithReadMessages.messagesWithBatch where sequence-number gaps (from hard deletes and soft-delete cleanup) could cause batched streams to prematurely stop (bounded queries) or poll the same empty window forever (live queries). It does so by replacing the previous windowing-only unfoldAsync logic with a small query-plan state machine that can safely fall back to full-range queries when a windowed batch is short, ensuring messages beyond gaps are still emitted.
Changes:
- Reworked
internalBatchStreaminto a query-plan state machine (QueryRemaining,QueryWindow,PollRemaining,Complete) to correctly handle gaps while preserving the dense fast-path windowing optimization. - Added a comprehensive in-memory unit test suite (
MessagesWithBatchTest) specifying correct behavior across a variety of gap/polling scenarios. - Added database contract tests (
MessagesWithBatchDatabaseContractTest) and integration-test wrappers to validate behavior across supported DBs (including hard/soft deletes, journaldelete, and deserialization failures).
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated no comments.
| File | Description |
|---|---|
| core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/BaseJournalDaoWithReadMessages.scala | Implements the new query-plan state machine to ensure gaps don’t truncate or stall batched reads. |
| core/src/test/scala/org/apache/pekko/persistence/jdbc/journal/dao/MessagesWithBatchTest.scala | In-memory state-machine specification covering gap shapes, polling, completion behavior, and failure propagation. |
| core/src/test/scala/org/apache/pekko/persistence/jdbc/journal/dao/MessagesWithBatchDatabaseContractTest.scala | DB-backed contract tests for delete semantics and corrupt-row failure behavior (H2 in core, reused by integration). |
| integration-test/src/test/scala/org/apache/pekko/persistence/jdbc/integration/MessagesWithBatchDatabaseContractTest.scala | Integration-test DB matrix wrappers to run the contract suite on Postgres/MySQL/MariaDB/Oracle/SQL Server. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
@mucciolo if you are working on a PR then there is no need for a separate issue. Issues are best used when there is a need to track something that is not yet ready to work on. |
He-Pin
left a comment
He-Pin
left a comment
There was a problem hiding this comment.
Code Review: Emit all messages across sequence number gaps
Overall Assessment
This is a well-crafted fix for a real correctness bug. The QueryPlan state machine is a clean, well-documented design that correctly handles sequence number gaps while preserving the performance safeguard of windowed queries from PR #180.
Findings
🟡 suggestion: LimitWindowingStreamTest query range expectations may be stale
LimitWindowingStreamTest.scala:77 calls internalBatchStream directly. The old code used limitWindow(from) which produced windows of batchSize sequence numbers (e.g., [101, 200] for batchSize=100). The new code uses windowedQuery(from) which produces windows of batchSize + 1 sequence numbers (e.g., [101, 201]). This is intentional — the extra slot lets a window tolerate one missing sequence number and still return a full batch.
The test only checks batchCount and totalCount, not queriedRanges, so for consecutive messages it still passes (both produce 1600 batches of 100). However, if the test is ever extended to check ranges, it will fail. Consider updating the test to also record and verify the new expected ranges, or adding a comment explaining the difference.
✅ Strengths
-
Correct gap handling: The
QueryRemaining→QueryWindow→QueryRemainingfallback on short windowed batches is the right design. A short windowed batch is inconclusive (gaps may hide messages), while a short full-range batch is conclusive (reached the end). -
First query full range: Starting with
QueryRemaininginstead of a windowed query correctly handles leading gaps (e.g., journal purged up to a snapshot) in a single round trip, subsuming the #195 special case. -
Overflow handling: The
windowedQueryoverflow guardfrom <= (Long.MaxValue - normalizedBatchSize)is correct and matches the oldlimitWindowlogic. -
Poll semantics preserved:
PollRemainingpolls the full remaining range, which is correct — a windowed poll can never reach messages appended beyond a trailing gap. -
Comprehensive test coverage: 26 unit tests covering narrow/wide gaps, leading/trailing gaps, live polling, bounded live completion, failure pass-through, and the two interaction tests pinning first-query full range and dense windowing.
-
normalizedBatchSize = Math.max(1, batchSize): Correctly handles batchSize=0 and negative values.
-
isEmptyRange = from > toSequenceNr: Correctly handles fromSequenceNr beyond toSequenceNr without polling forever.
-
FlowControl not deleted: The PR correctly removes only the import from this file; FlowControl is still used by JdbcDurableStateStore and JdbcReadJournal.
No blockers found. The implementation is correct and the test coverage is thorough.
|
|
||
| /** | ||
| * separate this method for unit tests. | ||
| * Separate this method for unit tests. |
There was a problem hiding this comment.
🟡 suggestion: The window size changed from batchSize to batchSize + 1 sequence numbers. This is correct — the extra slot lets a window tolerate one gap and still return a full batch. However, LimitWindowingStreamTest.scala:77 calls internalBatchStream directly. The old code produced windows of batchSize numbers (e.g., [101, 200]); the new code produces batchSize + 1 (e.g., [101, 201]). The test only checks batch/message counts (not ranges), so it still passes for consecutive messages. But if it's ever extended to check queriedRanges, it will fail. Consider updating it or adding a comment.
There was a problem hiding this comment.
Kept this count-only with a note on the batchSize+1 width. The exact ranges are already pinned in-memory by MessagesWithBatchTest, so asserting them here too would duplicate that and couple the test to the window arithmetic. 5d8c8e1
| val normalizedFromSequenceNr = Math.max(1, fromSequenceNr) | ||
|
|
||
| def pollOrComplete(from: Long): QueryPlan = refreshInterval match { | ||
| case Some((delay, scheduler)) => PollRemaining(from, delay, scheduler) |
There was a problem hiding this comment.
✅ good: Handling from > toSequenceNr as isEmptyRange correctly prevents a live stream from polling a range that can never produce messages. This edge case is important for the fromSequenceNr > toSequenceNr scenario (tested in MessagesWithBatchTest).
| QueryWindow(from, endInclusive) | ||
| } | ||
|
|
||
| def retrieveBatch(from: Long, endInclusive: Long): Future[Option[(QueryPlan, Seq[Try[(PersistentRepr, Long)]])]] = |
There was a problem hiding this comment.
✅ nice design: The unfoldAsync pattern matching on QueryPlan subtypes is much cleaner than the old FlowControl approach. The state machine is self-documenting — each case in the match clearly shows what query runs and what the next state is. The sealed trait hierarchy with QueryWindow/QueryRemaining/PollRemaining/Complete makes the transition graph explicit.
| private[dao] object BaseJournalDaoWithReadMessages { | ||
|
|
||
| /** The query the batch stream will run. */ | ||
| private sealed trait QueryPlan |
There was a problem hiding this comment.
"Query" is also a verb, which implies action, so no semantic gain over QueryPlan. Happy to consider alternatives if you feel strongly.
|
@mucciolo I added apache/pekko-persistence-r2dbc#398 to bring similar test coverage to pekko-persistence-r2dbc. |
) (#521) * Emit all messages across sequence number gaps in messagesWithBatch #516 * The query windowing introduced by #180 treated a windowed batch returning fewer messages than batchSize as the end of the journal: bounded streams (recovery, currentEventsByPersistenceId) completed early and silently dropped every message beyond a gap left by deleted messages, and live streams stalled forever polling an empty window. #195 fixed only the gap at the head of the journal. * Rework the unfoldAsync state from FlowControl into a QueryPlan state machine (QueryRemaining, QueryWindow, PollRemaining, Complete): a short windowed batch is no longer conclusive and falls back to one query over the full remaining range, which gaps cannot hide messages from. * Keep windowing as the dense fast path of #180: after a full batch shows the journal to be dense, queries stay bounded to [from, from + batchSize]. * Span the full remaining range on the first query, crossing a deleted journal head (snapshot cleanup) in a single round trip, subsuming the #195 special case. * Poll the full remaining range when tailing: a windowed poll can never reach messages appended beyond a trailing gap. * Specify the gap state machine in MessagesWithBatchTest (core, in-memory journal stub) and the database-coupled behavior in MessagesWithBatchDatabaseContractTest (H2 plus the five integration databases): hard and soft deletes, a prefix purge through the journal's delete, and serialization failures. * Use a non-empty writer in the corrupt-row test fixture Oracle treats the empty string as NULL, so inserting the synthetic corrupt row with writer = "" violated the NOT NULL constraint on EVENT_JOURNAL.WRITER (ORA-01400) and failed only the Oracle integration tests. The production write path always carries a writer UUID, so this is a test-fixture-only fix. * Use the standard Apache header on the new test files These files contain no Akka-derived code, so they should carry the full ASF header (the headerLicense the build stamps on new files) rather than the shorter "derived from Akka" variant, per a maintainer review note. * Note batchSize+1 window width in LimitWindowingStreamTest Co-authored-by: Diego D. Mucciolo <2973680+mucciolo@users.noreply.github.com>
And I to Persistence TCK apache/pekko#3076 |
Fixes #516
The bug
messagesWithBatchbackseventsByPersistenceId,currentEventsByPersistenceIdand actor recovery. Since the query windowing introduced in #180, a batch window[from, from + batchSize]containing fewer live messages thanbatchSizeis treated as the end of the journal:Gaps in sequence numbers are produced by
deleteMessages(hard deletes below the retained marker row) and by cleanup of soft-deleted rows. #195 fixed this only for a gap at the head of the journal (the snapshot cleanup case). Upstream akka-persistence-jdbc does not have this bug. A full analysis and regression timeline is in #516.The fix
The
unfoldAsyncstate inBaseJournalDaoWithReadMessagesbecomes a small query-plan state machine:QueryRemaining(from)queries the whole remaining range[from, toSequenceNr]with the batch-size LIMIT. It is planned first, and whenever a short windowed batch leaves the remainder undetermined, since gaps cannot hide messages from an unwindowed query.QueryWindow(from, endInclusive)is the dense fast path, planned only after a full batch showed the journal to be dense. This keeps the perf safeguard of perf: avoid large offset query via limit windowing #180; it is not load-bearing for correctness because a short window falls back toQueryRemaining.PollRemaining(from, delay, scheduler)is the live tail. It polls the full remaining range: a windowed poll can never reach messages appended beyond a trailing gap.Complete.flowchart TD start((start)) --> QueryRemaining QueryRemaining -->|"full batch"| QueryWindow QueryRemaining -->|"short batch, polling"| PollRemaining QueryRemaining -->|"reached the end, or short batch when not polling"| Complete QueryWindow -->|"full batch"| QueryWindow QueryWindow -->|"short batch before toSequenceNr (gap fallback)"| QueryRemaining QueryWindow -->|"short batch at toSequenceNr, polling"| PollRemaining QueryWindow -->|"reached the end, or short batch at toSequenceNr when not polling"| Complete PollRemaining -->|"full batch"| QueryWindow PollRemaining -->|"short batch"| PollRemaining PollRemaining -->|"reached the end"| Complete Complete --> done((done)) linkStyle 5 stroke:red,stroke-width:2px,color:redbatchSizemessages; short batch: fewer.toSequenceNr(or, on the first query, the requested range was already empty).The bug was the missing
QueryWindowtoQueryRemainingedge, drawn in red above: a short windowed batch was treated as the end of the journal.Notes:
Tests
MessagesWithBatchTest(core): the gap state machine specified against an in-memory journal stub honoring themessagescontract (ascending order, inclusive bounds, LIMIT). 26 tests covering gaps narrower and wider than the batch size, leading and trailing gaps, live polling across gaps, bounded live completion, failure pass-through and the two interaction tests pinning the first-query full range and the dense windowing of perf: avoid large offset query via limit windowing #180.MessagesWithBatchDatabaseContractTest(H2 in core; Postgres, MySQL, MariaDB, Oracle and SQL Server in integration-test): hard-deleted gaps, soft-deleted messages, a mixed hard/soft gap, a prefix purge through the journal'sdelete, and a corrupt-row serialization failure.