pjfanning commented on code in PR #3653: URL: https://github.com/apache/pekko/pull/3653#discussion_r4358369064
########## persistence/src/test/scala/org/apache/pekko/persistence/WriteMessagesSpec.scala: ########## @@ -0,0 +1,72 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.pekko.persistence + +import org.apache.pekko +import pekko.actor.ActorRef +import pekko.persistence.JournalProtocol.WriteMessages + +import org.scalatest.matchers.should.Matchers +import org.scalatest.wordspec.AnyWordSpec + +class WriteMessagesSpec extends AnyWordSpec with Matchers { + + private val writes = AtomicWrite(PersistentRepr("a", 1, "p1")) :: Nil + + "JournalProtocol.WriteMessages" must { + "not bypass the circuit breaker by default" in { + WriteMessages(writes, ActorRef.noSender, 1).bypassCircuitBreaker should ===(false) + new WriteMessages(writes, ActorRef.noSender, 1).bypassCircuitBreaker should ===(false) + WriteMessages(writes, ActorRef.noSender, 1, bypassCircuitBreaker = true).bypassCircuitBreaker should ===(true) + } + + "keep bypassCircuitBreaker in copy" in { + val w = WriteMessages(writes, ActorRef.noSender, 1, bypassCircuitBreaker = true) + val copied = w.copy(actorInstanceId = 2) + copied.actorInstanceId should ===(2) + copied.bypassCircuitBreaker should ===(true) + w.withBypassCircuitBreaker(false).bypassCircuitBreaker should ===(false) + } + + "extract the first 3 fields in pattern matching" in { + val w: Any = WriteMessages(writes, ActorRef.noSender, 3, bypassCircuitBreaker = true) + w match { + case WriteMessages(messages, persistentActor, actorInstanceId) => + messages should ===(writes) + persistentActor should ===(ActorRef.noSender) + actorInstanceId should ===(3) + case other => + fail(s"unexpected $other") + } + } + + "include bypassCircuitBreaker in equality" in { + WriteMessages(writes, ActorRef.noSender, 1) should ===(WriteMessages(writes, ActorRef.noSender, 1)) + WriteMessages(writes, ActorRef.noSender, 1).hashCode should ===( Review Comment: Fixed the underlying contract in fd68ca81ad. `PersistentImpl.hashCode` now hashes a null sender (`ActorRef.noSender`) as 0; the hash is unchanged for other senders. `HashCode.hash` itself still rejects null, because `HashCodeSpec` deliberately pins that behaviour since #3312. `WriteMessagesSpec` gets a `PersistentRepr` hashCode test without a sender. It and the equality test fail against the previous code and pass now. ########## persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/EventWriter.scala: ########## @@ -378,6 +612,58 @@ private[pekko] object EventWriter { } Behaviors.same + case MaxSeqNrForPid(pid, maxSeqNr, reason @ (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)) + waiting.foreach { + case (repr, replyTo) => + handleWrite( + repr, + replyTo, + isSnapshotEvent = reason == AskMaxSeqNrReason.SnapshotEvent, + calledAfterMaxSeqNr = true) + } + } + Behaviors.same + + case MaxSeqNrForPidFailed(pid, errorDescOpt) => + // journal seems to have real problems, let the circuit breaker protect it until next successful write + bypassCircuitBreaker = false + def errorDescInLog = errorDescOpt match { + case AskMaxSeqNrReason.WriteFailure(errorDesc) => s", original error desc: $errorDesc" + case _ => "" + } + perPidWriteState.get(pid) match { + case None => + context.log.debug2( + "Got max seq nr (failed) with no waiting previous state for persistence id [{}]{}", + pid, + errorDescInLog) + case Some(state) => + state.waitingForReply.values.foreach { + case (_, replyTo) => + replyTo ! StatusReply.error("Journal write failed") + } + val sortedSeqs = state.waitingForReply.keys.toSeq.sorted Review Comment: Fixed in fd68ca81ad. `MaxSeqNrForPidFailed` now carries the cause, from either `ReplayMessagesFailure` or the ask failure. It fails every write in `waitingForReply` and `waitingForSeqNrLookup` with `Journal write failed`, with the cause attached, and clears both queues. The log no longer calls `head`/`last`. New tests in `EventWriterFillGapsSpec` cover an initial lookup failure and an ask timeout, each followed by a successful later write. Both fail against the previous code. ########## persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/EventWriter.scala: ########## @@ -378,6 +612,58 @@ private[pekko] object EventWriter { } Behaviors.same + case MaxSeqNrForPid(pid, maxSeqNr, reason @ (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)) + waiting.foreach { + case (repr, replyTo) => + handleWrite( + repr, + replyTo, + isSnapshotEvent = reason == AskMaxSeqNrReason.SnapshotEvent, Review Comment: Fixed in fd68ca81ad. Queued lookups are now `PendingLookup(repr, replyTo, isSnapshotEvent)`, and each write is replayed with its own flag. New `EventWriterSpec` regression with gap filling off: a snapshot event at 5 and an ordinary event at 7 are queued for one lookup (highest 3). The journal now receives `[4, 5]` and then `[7]`. Before, 6 was filled as well. ########## persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/EventWriter.scala: ########## @@ -269,8 +417,124 @@ private[pekko] object EventWriter { 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((repr, replyTo))) + } 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 :+ ((repr, replyTo))) + } 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) { + context.log.trace2("Duplicate seq nr [{}] for persistence id [{}]", sequenceNumber, persistenceId) + replyTo ! StatusReply.success(WriteAck(persistenceId, sequenceNumber)) // duplicate Review Comment: Fixed in fd68ca81ad. When a duplicate's sequence number is still pending in `waitingForReply` or `waitingForWrite`, its caller is kept in `duplicateWaiters` and gets the same outcome as that write: success, failure (including after the write-failure lookup) or rejection. Only sequence numbers that are no longer pending are acknowledged right away. New tests check that the duplicate gets no reply while the write is unresolved, then a success or an error matching the journal outcome. -- 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]
