pjfanning commented on PR #489:
URL:
https://github.com/apache/pekko-persistence-r2dbc/pull/489#issuecomment-5571471396
Thanks for this — the core design is sound. The two hard requirements
(`use-app-timestamp`, `db-timestamp-monotonic-increasing`) are exactly what
makes cross-pid batching safe: with `dbTimestampMonotonicIncreasing = on`,
`JournalDao.writeEvents` never binds the per-pid `previousSeqNr` subselect, so
mixing persistence ids in one statement is legitimate. `updateInBatchReturning`
goes through `withConnection`, which is transactional, so the bisection retry
can't double-write. And `publish` correctly uses each row's own app timestamp,
so pub-sub offsets match the stored `db_timestamp`.
### What I verified locally
| Check | Result |
|---|---|
| `core/Test/compile` (2.13.18) | pass |
| `core/Test/compile` (3.3.8) | pass |
| `headerCheckAll` / `scalafmtCheckAll` | pass |
| Deprecation warnings | 1 new (A3 below) |
| Tests | **not run** — need a live Postgres |
MiMa is a non-issue here: everything new is `private[r2dbc]` /
`@InternalApi`, and the `R2dbcSettings.scala` change is comment-only.
## A. Should be addressed before merge
**A1. The queue is unbounded — this was an explicit requirement in #471.**
`R2dbcBatchJournal.scala:127`. The issue says *"The buffer is bounded; on
overflow new writes are rejected (the originating persistent actor sees the
write fail) rather than blocking"*, and open question 4 asked us to pick that
contract. The PR ships neither a bound nor rejection — the docs section states
the queue is unbounded as a fact. If the database stalls, every serialized
payload accumulates on the heap with nothing to stop it. The default journal is
bounded in practice by the connection pool; this one isn't bounded by anything.
**A2. A `publish` failure can mark committed writes as failed.**
`R2dbcBatchJournal.scala:138-151`:
```scala
.map { _ =>
requests.foreach { w =>
publish(w.messages, Future.successful(w.rows.head.dbTimestamp))
w.promise.trySuccess(Nil)
}
}
.recoverWith { ... case exception =>
requests.foreach(_.promise.tryFailure(exception)) ... }
```
If `publish` throws for request *i*, requests *i+1..n* never reach
`trySuccess`, and the catch-all in `recoverWith` fails them — even though their
rows are committed. Those actors get `WriteMessagesFailed` and stop, then
replay events that are in the journal. Low probability (`PubSub.publish`
throwing is unlikely), but trivially avoided: complete all promises first, then
publish in a second pass.
**A3. Deprecated `Array` -> `Seq` copy in the flush hot path.**
`R2dbcBatchJournal.scala:137` — confirmed warning on both 2.13 and 3:
> `method copyArrayToImmutableIndexedSeq ... is deprecated (since 2.13.0):
implicit conversions from Array to immutable.IndexedSeq are implemented by
copying`
`requests.flatMap(_.rows)` builds an `Array` and then silently copies it
again into a `Seq` on every flush, in the one code path this feature exists to
make fast. Use a `Vector`/`ArraySeq` for `writeRequests` instead of
`Array.fill`, or `ArraySeq.unsafeWrapArray`.
**A4. No dialect check.** #471 specified *"Requirements checked at startup:
`use-app-timestamp = on`, `db-timestamp-monotonic-increasing = on`, and the
postgres or yugabyte dialect"*. The two timestamp requirements are there
(`R2dbcBatchJournal.scala:110-113`); the dialect one isn't. MySQL satisfies
both timestamp requirements by construction (see
`MySQLJournalDao.settingRequirements`), so a MySQL user can enable this today
into a path that is untested and whose `insertEventWithParameterTimestampSql`
has no `RETURNING`. Either add the `require`, or test against MySQL and
document it as supported.
**A5. No separate plugin id — enabling this rewires the default journal.**
#471 proposed `pekko.persistence.r2dbc.batched-journal` as its own plugin id.
This PR instead has users override `pekko.persistence.r2dbc.journal.class`.
That means the default and batched journals can't run side by side in one
system (cf. `MultiPluginSpec`), and it mutates a config block that also feeds
the docs snippet. A `batched-journal` block in `reference.conf` inheriting from
`journal` would be cleaner and matches the issue.
## B. Worth fixing
**B1. Dropped `Flush` messages become dead letters.**
`R2dbcBatchJournal.scala:158` — `case Flush if noActiveWrite &&
queue.nonEmpty`. When the guard is false the message isn't matched by the
composed `Receive`, so it goes to `unhandled()`, which publishes
`UnhandledMessage` and gets logged by `DeadLetterListener` (up to
`log-dead-letters`). This happens routinely — any timer firing while a flush is
in progress. Prefer `case Flush => if (noActiveWrite && queue.nonEmpty) { ...
}`. The scheduling logic itself looks correct as far as I can trace it: every
dropped `Flush` is recovered by the `FlushDone` branch. It's only the logging
that's wrong.
**B2. `max-batch-size` bounds requests, not rows.** One request can carry an
entire `persistAll`, so a flush of 100 requests can produce an arbitrarily
large multi-row insert. There is no cap on total rows or payload bytes per
statement. The docs note a request can hold several events but don't draw that
conclusion.
**B3. Bisection cost on a batch-wide failure.** If an
`R2dbcDataIntegrityViolationException` affects all requests (say a `NOT NULL`
violation), the halving performs ~2n-1 statements before failing everything —
roughly 199 round trips at the default `max-batch-size = 100`. Worth a depth
cap or a short-circuit.
**B4. Settings read with raw config accessors.**
`R2dbcBatchJournal.scala:115-116` uses `config.getInt`/`getDuration` directly
rather than going through `JournalSettings`, unlike every other setting in this
plugin.
**B5. The new keys leak into the default journal's documented settings.**
They're added to the `journal` block inside the `// #journal-settings` snippet
region, so they now render in the default journal's config table in
`journal.md`, where they do nothing.
## C. Minor
- `R2dbcBatchJournalFailureIsolationSpec.scala:61` — `private final case
class StoredRow` in the class body produces *"The outer reference in this type
test cannot be checked at run time"*. Move it to the companion object.
- `R2dbcBatchJournal.scala` duplicates ~150 lines verbatim from
`R2dbcJournal.scala` (`deserializeRow`, the whole serialization block,
`publish`, `asyncReplayMessages`, `asyncReadHighestSequenceNr`,
`asyncDeleteMessagesTo`, `WriteFinished`). That's two copies to keep in sync
forever; a shared base trait would be worth the refactor.
- `postStop` fails the queued promises, whose `onComplete` then sends
`WriteFinished` to the stopped actor — more dead letters.
- The `use-app-timestamp` comment rewrite drops the link to
yugabyte/yugabyte-db#10995 from both `reference.conf` and
`R2dbcSettings.scala`. Reasonable now that the setting is load-bearing for
MySQL and this plugin, but it's an unrelated change and the issue link is worth
keeping.
- The `db-timestamp-monotonic-increasing` `require` has no matching case in
`R2dbcBatchJournalValidationSpec` (only `max-batch-size` and
`use-app-timestamp` are covered).
## Tests
Coverage is good and the specs are directional — size-triggered flush,
timer-triggered flush, failure isolation through bisection with a DB-level
assertion that only the offending pid is missing, and per-request publish
timestamps. `R2dbcBatchJournalSpec` runs the TCK. Gaps: no
overload/backpressure test (nothing to test until A1 lands), no MySQL/Yugabyte
run, and nothing covering restart with a non-empty queue.
--
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]
