He-Pin commented on code in PR #3653:
URL: https://github.com/apache/pekko/pull/3653#discussion_r4359711171
##########
persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/EventWriter.scala:
##########
@@ -378,6 +687,62 @@ private[pekko] object EventWriter {
}
Behaviors.same
+ case MaxSeqNrForPid(pid, maxSeqNr, AskMaxSeqNrReason.FillGaps |
AskMaxSeqNrReason.SnapshotEvent) =>
+ bypassCircuitBreaker = true
+ perPidWriteState.get(pid) match {
+ case None =>
+ context.log.debug("Got max seq nr with no waiting previous
state for pid, ignoring (pid [{}])", pid)
+ case Some(state) =>
+ context.log.debug("Retrieved max seq nr [{}] for pid [{}]",
maxSeqNr, pid)
+ val waiting = state.waitingForSeqNrLookup
+ perPidWriteState = perPidWriteState.updated(
+ pid,
+ state.copy(latestSeqNr = maxSeqNr, waitingForSeqNrLookup =
Vector.empty))
+ // each queued write keeps its own snapshot event flag, the
lookup reason only
+ // belongs to the write that triggered it
+ waiting.foreach { pending =>
+ handleWrite(pending.repr, pending.replyTo,
pending.isSnapshotEvent, calledAfterMaxSeqNr = true)
+ }
Review Comment:
Please run the idle-state cleanup after replaying this lookup queue. With
gap filling enabled and cache capacity 2, send sequence 5 for six distinct
persistence ids and answer each initial lookup with highest=5. Every caller is
acknowledged as a duplicate, but all six states remain in perPidWriteState:
this path never calls handleUpdatedStateForPid, where eviction happens. Sending
sequence 6 for all six ids before acknowledging any writes produces six direct
WriteMessages instead of at most two. I reproduced this against fd68ca81ad. A
duplicate-only stream can therefore retain unbounded pid state despite
latest-sequence-number-cache-capacity. Please route the final idle state
through cleanup/eviction and add this lookup-only regression.
##########
persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/EventWriter.scala:
##########
@@ -264,13 +449,168 @@ private[pekko] object EventWriter {
}
Behaviors.same
+ case JournalProtocol.WriteMessageRejected(message, cause,
transactionId) =>
+ val pid = message.persistenceId
+ val sequenceNr = message.sequenceNr
+ perPidWriteState.get(pid) match {
+ case Some(state)
+ if transactionId == state.currentTransactionId &&
state.waitingForReply.contains(sequenceNr) =>
+ // the journal rejected the write before storing it, e.g. it
could not serialize the event
+ // or a FilteredPayload gap filler, so all events of the
atomic write are rejected
+ context.log.warnN(
+ "Journal rejected write of event persistence id [{}],
sequence nr [{}]: {}",
+ pid,
+ sequenceNr,
+ cause.getMessage)
+ val newState = state.replyTo(sequenceNr,
StatusReply.error(journalWriteRejected(cause)))
+ handleUpdatedStateForPid(pid, newState.copy(waitingForReply
= newState.waitingForReply - sequenceNr))
+ case _ =>
+ context.log.debugN(
+ "Got write rejection for event with no waiting request,
ignoring (pid [{}], seq nr [{}], tx id [{}])",
+ pid,
+ sequenceNr,
+ transactionId)
+ }
+ Behaviors.same
+
case _ =>
// ignore all other journal protocol messages
Behaviors.same
}
+ def handleWrite(
+ repr: PersistentRepr,
+ replyTo: ActorRef[StatusReply[WriteAck]],
+ isSnapshotEvent: Boolean,
+ calledAfterMaxSeqNr: Boolean): Unit = {
+ val persistenceId = repr.persistenceId
+ val sequenceNumber = repr.sequenceNr
+ val newStateForPid =
+ perPidWriteState.get(persistenceId) match {
+ case None =>
+ if ((fillSequenceNumberGaps || isSnapshotEvent) &&
sequenceNumber != 1L) {
+ val reason = if (isSnapshotEvent)
AskMaxSeqNrReason.SnapshotEvent else AskMaxSeqNrReason.FillGaps
+ askMaxSeqNr(persistenceId, reason)
+ StateForPid(
+ waitingForReply = Map.empty,
+ waitingForSeqNrLookup = Vector(PendingLookup(repr,
replyTo, isSnapshotEvent)))
+ } else {
+ sendToJournal(1, Vector(repr))
+ StateForPid(Map((repr.sequenceNr, (repr, replyTo))),
currentTransactionId = 1)
+ }
+
+ case Some(state) =>
+ val expectedSeqNr =
+ if (fillSequenceNumberGaps || isSnapshotEvent)
state.nextExpectedSeqNr
+ else repr.sequenceNr
+
+ if (state.seqNrlookupInProgress) {
+ context.log.trace2(
+ "Seq nr lookup in progress for persistence id [{}], adding
sequence nr [{}] to pending",
+ persistenceId,
+ sequenceNumber)
+ state.copy(waitingForSeqNrLookup =
+ state.waitingForSeqNrLookup :+ PendingLookup(repr,
replyTo, isSnapshotEvent))
+ } else if (sequenceNumber == expectedSeqNr) {
+ if (state.idle) {
+ sendToJournal(state.currentTransactionId + 1, Vector(repr))
+ state.copy(
+ waitingForReply = Map((repr.sequenceNr, (repr,
replyTo))),
+ currentTransactionId = state.currentTransactionId + 1)
+ } else {
+ // write in progress for pid, add write to batch and
perform once current write completes
+ if
(state.waitingForWriteExceedingMaxBatchSize(settings.maxBatchSize)) {
+ replyTo ! StatusReply.error(
+ s"Max batch reached for pid $persistenceId, at most
${settings.maxBatchSize} writes for " +
+ "the same pid may be in flight at the same time")
+ state
+ } else {
+ context.log.trace2(
+ "Writing event in progress for persistence id [{}],
adding sequence nr [{}] to batch",
+ persistenceId,
+ sequenceNumber)
+ state.copy(waitingForWrite = state.waitingForWrite :+
((repr, replyTo)))
+ }
+ }
+ } else if (sequenceNumber < expectedSeqNr) {
+ if (state.isPending(sequenceNumber)) {
+ // not durable yet, reply when the outcome of the pending
write is known
+ context.log.trace2(
+ "Duplicate of pending seq nr [{}] for persistence id
[{}]",
+ sequenceNumber,
+ persistenceId)
+ state.copy(duplicateWaiters =
state.duplicateWaiters.updated(
+ sequenceNumber,
+ state.duplicateWaiters.getOrElse(sequenceNumber,
Vector.empty) :+ replyTo))
+ } else {
+ context.log.trace2("Duplicate seq nr [{}] for persistence
id [{}]", sequenceNumber, persistenceId)
+ replyTo ! StatusReply.success(WriteAck(persistenceId,
sequenceNumber)) // duplicate
+ state
+ }
+ } else { // sequenceNumber > expectedSeqNr
+ require(
+ fillSequenceNumberGaps || isSnapshotEvent,
+ s"Unexpected sequence number gap, expected
[$expectedSeqNr], received [$sequenceNumber]. " +
+ "Enable
pekko.persistence.typed.event-writer.fill-sequence-number-gaps config if gaps
are " +
+ "expected and should be filled with FilteredPayload.")
+
+ if (calledAfterMaxSeqNr || !state.idle) {
+ // here we can trust that the expectedSeqNr is correct and
don't have to lookup max
+ val fillRepr = (expectedSeqNr until sequenceNumber).map {
n =>
+ PersistentRepr(
+ FilteredPayload,
+ persistenceId = persistenceId,
+ sequenceNr = n,
+ manifest = "", // adapters would be on the producing
side, already applied
+ writerUuid = writerUuid,
+ sender = pekko.actor.ActorRef.noSender)
+ }.toVector
+ context.log.debugN(
+ "Detected sequence nr gap{} [{}-{}] for persistence id
[{}]. Filling with FilteredPayload.",
+ if (isSnapshotEvent) " for snapshot event" else "",
+ fillRepr.head.sequenceNr,
+ fillRepr.last.sequenceNr,
+ persistenceId)
+ val ignoreRef = context.system.ignoreRef
+ if (state.idle) {
+ sendToJournal(state.currentTransactionId + 1, fillRepr
:+ repr)
+ val newWaitingForReply =
+ (fillRepr.map(r => r.sequenceNr -> (r -> ignoreRef)) :+
+ (repr.sequenceNr -> (repr -> replyTo))).toMap
+ state.copy(
+ waitingForReply = newWaitingForReply,
+ currentTransactionId = state.currentTransactionId + 1)
+ } else {
+ if (context.log.isTraceEnabled)
+ context.log.traceN(
+ "Writing event in progress for persistence id [{}],
adding sequence nrs [{}-{}] to batch",
+ persistenceId,
+ fillRepr.head.sequenceNr,
+ sequenceNumber)
+ val newWaitingForWrite = fillRepr.map(_ -> ignoreRef) :+
(repr -> replyTo)
+ state.copy(waitingForWrite = state.waitingForWrite ++
newWaitingForWrite)
Review Comment:
This gap branch appends another real event without checking maxBatchSize.
With gap filling enabled and max-batch-size=2, hold sequence 1 in flight and
enqueue 3, 5, then 7. After acknowledging 1, the next AtomicWrite contains real
events [3,5,7] plus fillers; I reproduced this against fd68ca81ad. With gap
filling disabled, the same sequence correctly rejects 7 and batches [3,5].
Repeating gapped inputs can grow the pending batch without the configured
backpressure. Please apply the real-event batch-limit check before
creating/appending the fillers, and test gapped writes at the limit.
--
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]
