pjfanning commented on PR #489:
URL:
https://github.com/apache/pekko-persistence-r2dbc/pull/489#issuecomment-5791200441
Thanks for this. I rebased the branch onto current `main` (no conflicts, the
PR diff is unchanged) so it picks up the recent workflow/dependency updates.
I went through the new journal. The parts I was most worried about hold up:
the flush/timer state machine has no stall or double-flush path in steady
state, `doFlush` can't be entered with an empty queue, the bisection recursion
always shrinks and terminates, `updateInBatchReturning` really is transactional
so a failed mixed-pid batch rolls back fully, and `JournalDao.writeEvents`
genuinely doesn't bind the per-pid `previousSeqNr` when
`db-timestamp-monotonic-increasing = on`, so mixing persistence ids in one
statement is safe as claimed.
A few things I'd like addressed:
**1. One `db_timestamp` per flush can break `eventsBySlices` (the one I'd
fix before merge)**
`R2dbcBatchJournal.scala:161` takes `InstantFactory.now()` once per flush
(truncated to microseconds) and stamps it on every row.
`BySliceQuery.deserializeAndAddOffset` (`internal/BySliceQuery.scala:536`)
throws `IllegalStateException("Too many events stored with the same timestamp
... buffer size [1000]")` once `buffer-size` rows in a queried slice range
share a timestamp.
`max-batch-size` bounds *requests* (default 100), not events — and the
scaladoc and docs both say a request may carry an arbitrary number of events
via `persistAll`/`persistAsync`. So 100 entities each persisting 15 events
inside one 10 ms window produces 1500 rows on a single `db_timestamp`, and
every live `eventsBySlices` over those slices fails and the projection stops.
The default journal can't hit this because each `asyncWriteMessages` call gets
its own `now()`.
Could we cap rows per flush (split when `rows.size` approaches
`buffer-size`), or spread microseconds across the batch?
**2. Timestamp is taken before the connection is acquired**
Same line. `journalDao.writeEvents` only then goes through
`R2dbcExecutor.getConnection`, whose `acquire-timeout` defaults to 5s
(`reference.conf:259`), while `behind-current-time` defaults to 100ms
(`reference.conf:343`). Under pool contention a row commits well after its
`db_timestamp`, so a live `eventsBySlices` that already advanced past it skips
the event until backtracking catches it ~10s later. `journal.md` tells the user
to raise `behind-current-time`, but the shipped `batched-journal` block
doesn't, so the default config is the one exposed. The default journal
sidesteps this with `transaction_timestamp()`.
**3. A stale `FlushDone` can survive an actor restart**
`postStop` (line 328) fails and clears the queue, but the already-dequeued
in-flight batch keeps running and its `onComplete` still does `self !
FlushDone` (line 181). `asyncWriteMessages` is shielded by
`AsyncWriteJournal`'s try/catch, but `receivePluginInternal` isn't — so on a
restart the new instance can already have a flush in flight when the stale
`FlushDone` arrives, set `noActiveWrite = true` and call `doFlush()`, giving
two concurrent flushes. That breaks the same-persistence-id ordering guarantee
documented at lines 94-97. Timer messages are protected from exactly this by
`TimerSchedulerImpl`'s `owner ne this` check; a generation/instance token on
`FlushDone` would do the same.
**4. Mixed-pid precondition is documented but not enforced**
`JournalDao.scala:177-181` now says mixed persistence ids are allowed, but
`writeEvents` has no `require(settings.dbTimestampMonotonicIncreasing)`. In the
`else` branch (line 231+) the per-row `persistenceId` is paired with
`previousSeqNr` derived from `events.head`, so a mixed batch would feed
mismatched `(pid, seqNr)` pairs to the `GREATEST(?, subselect)` timestamp
subselect and silently compute wrong timestamps. It's unreachable today only
because `R2dbcBatchJournal`'s constructor `require`s stop it — a `require` here
would make the contract enforced rather than documented.
**5. `publish` failures are silently swallowed**
`R2dbcBatchJournal.scala:171` — `publish(requests, dbTimestamp)` runs inside
the `.map` covered by `recoverWith`, after `promise.trySuccess(Done)`. If
`ps.publish` throws part-way through the loop the remaining requests are never
published, and the `case exception` branch's `tryFailure` is a no-op on
already-completed promises, so the batch ends as `Future.unit` with no log and
no signal. Worth moving `publish` outside the `recoverWith` or giving it its
own try/log.
**6. `max-batch-time` is unvalidated**
`R2dbcBatchJournal.scala:133` — `max-queue-size` and `max-batch-size` both
get a `require` on the following lines, but `max-batch-time` doesn't. `0ms` (or
negative) makes `startSingleTimer` fire immediately, degenerating into one
flush per request with all the batching overhead and none of the benefit, and
no diagnostic. A matching `require(maxBatchTime > Duration.Zero, ...)` would
round it out.
(1) is the blocker for me; the rest are hardening.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
