[leviramsey]

The JournalProtocol used by persistent actors (typed and classic) requires an ordering per-actor of responses to persists matching the order of those persists. Typed and classic persistent actors ensure that ordering holds by not issuing another WriteMessages while a WriteMessages is still outstanding. The AsyncWriteJournal in turn ensures this order by using a stronger ordering at journal level, covering all persistent actors using a particular journal. This ordering is enforced by the Resequencer actor.

The Resequencer's enforcement of a per-journal ordering can create a head-of-line-blocking followed by a thundering-herd if there are persists which take sufficiently abnormally long relative to more normal persists started immediately afterwards, especially if the amount by which the outliers deviate exceeeds the usual time between persists from different actors. For example, if most persists take 10ms and there are usually 100 distinct actors persisting events in a second (so persists come, on average, every 10ms), a persist taking 500ms would be expected to have ~50 actors whose persists succeeded but which won't report success until the 500ms persist completes (injecting an effective delay of ~250ms on average). The 50 responses would then be sent out within a few microseconds, with each response going to a different mailbox, with a likely surge in dispatcher utilization followed by (for the persistent actors send replies after persist across the cluster) a spike in messages being sent through remoting.

This change introduces the ability to configure the number of resequencers to use. The UID of the persistent actor which sent WriteMessages is used to consistently route messages for that actor via the same resequencer. Actors assigned to the same resequencer may still cause head-of-line-blocking for each other, but the impact of outliers is limited to only that subset of actors: actors using other resequencers are unaffected.

The default config is no change from prior behavior. To the extent that an occasional outlier is a symptom of an overloaded persistence backend, the resequencer introduces a small amount of backpressure on persists by delaying the next persist for an actor waiting for the head-of-line to unblock, so some caution would be advised in increasing this.

[leviramsey]

Benchmarked with a pathological in-memory store (high variation in persist latency) with 1000 actors persisting ~100 events each as fast as they can:

  // with single resequencer (before):    
  //  Persisted 101000 events (via 1000 actors) in 876 seconds    
  //  [115 persists/second]    
  // with 2 resequencers:    
  //  Persisted 101000 events (via 1000 actors) in 829 seconds    
  //  [121 persists/second]    
  // with 4 resequencers:    
  //  Persisted 101000 events (via 1000 actors) in 785 seconds    
  //  [128 persists/second]    
  // with 8 resequencers:    
  //  Persisted 101000 events (via 1000 actors) in 744 seconds    
  //  [135 persists/second]    
  // with 16 resequencers:    
  //  Persisted 101000 events (via 1000 actors) in 687 seconds    
  //  [147 persists/second]    
  // with 32 resequencers:    
  //  Persisted 101000 events (via 1000 actors) in 642 seconds    
  //  [157 persists/second]    
  // with 64 resequencers:    
  //  Persisted 101000 events (via 1000 actors) in 564 seconds    
  //  [179 persists/second]    
  // with 128 resequencers:    
  //  Persisted 101000 events (via 1000 actors) in 516 seconds    
  //  [195 persists/second]    
  // with 256 resequencers:    
  //  Persisted 101000 events (via 1000 actors) in 484 seconds    
  //  [208 persists/second]    
  // with 512 resequencers:    
  //  Persisted 101000 events (via 1000 actors) in 437 seconds    
  //  [231 persists/second]    
  // with 1024 resequencers:    
  //  Persisted 101000 events (via 1000 actors) in 385 seconds    
  //  [262 persists/second]    
  // with 2048 resequencers:    
  //  Persisted 101000 events (via 1000 actors) in 377 seconds    
  //  [267 persists/second]    

Benchmark:

 class DelayingInMemPlugin extends AsyncWriteJournal with AsyncRecovery {
    // Instantaneous recovery, a miracle
    override def asyncReadHighestSequenceNr(persistenceId: String, fromSequenceNr: Long): Future[Long] = zeroFuture
  
    override def asyncReplayMessages(persistenceId: String, fromSequenceNr: Long, toSequenceNr: Long, max: Long)(
        recoveryCallback: PersistentRepr => Unit): Future[Unit] =
      Future.unit
  
    override def asyncWriteMessages(messages: Seq[AtomicWrite]): Future[Seq[Try[Unit]]] = {
      if (schedulerGranularityMillis < 1) {    
        schedulerGranularityMillis =    
          context.system.settings.config.getDuration("akka.scheduler.tick-duration").toMillis.min(Int.MaxValue / 2).toInt    
      }    
      
      val rand = ThreadLocalRandom.current()    
      val delayMillis = {    
        val first = rand.nextInt(101)    
        first * rand.nextInt(first + 1)    
      }    
      val fut = Future.successful(messages.map(_ => successUnit))    
      
      if (delayMillis < schedulerGranularityMillis && rand.nextInt(schedulerGranularityMillis) > delayMillis) fut    
      else after(delayMillis.millis)(fut)(context.system)    
    }    
      
    override def asyncDeleteMessagesTo(persistenceId: String, toSequenceNr: Long): Future[Unit] = Future.unit    
      
    val zeroFuture = Future.successful(0L)    
    val successUnit = Success(())    
    var schedulerGranularityMillis: Int = -1    
  }    

 object Benchmark {
    val config = ConfigFactory.parseString("""
      akka.persistence.journal.plugin = delaying-in-mem
  
      delaying-in-mem {
        class = "akka.persistence.journal.DelayingInMemPlugin"
        num-resequencers = 2048
        circuit-breaker.call-timeout = 11s
      }
      """).withFallback(ConfigFactory.load())
  
    def main(args: Array[String]): Unit = {
      implicit val system = ActorSystem("Resequencer-Benchmark", config)
      import system.dispatcher
  
      val totalEvents = if (args.length > 0) args(0).toInt else 100000
      val numActors = if (args.length > 1) args(1).toInt.min(totalEvents) else 1000
      val eventsPerActor = 1 + (totalEvents / numActors)
  
      val actors = (1 to numActors).iterator
        .map(i => system.actorOf(testPersistentActor(s"test-persistent-$i"), s"test-persistent-$i"))
        .toArray
  
      implicit val timeout: Timeout = 20.seconds
  
      val streams = actors.iterator.map { actor =>
        Source
          .repeat(())
          .take(eventsPerActor)
          .mapAsync(1) { _ =>
            actor ? "persist"
          }
          .toMat(Sink.ignore)(Keep.right)
      }.toSeq
  
      system.log.info("Using {} resequencers", system.settings.config.getInt("delaying-in-mem.num-resequencers"))
  
      val startTime = System.nanoTime()
  
      Await.result(
        Future.sequence(streams.iterator.map(_.run()).toSeq).flatMap { _ =>
          val endTime = System.nanoTime()
          val elapsed = endTime - startTime
  
          system.log.info(
            "Persisted {} events (via {} actors) in {} seconds",
            eventsPerActor * numActors,
            numActors,
            elapsed / 1000000000L)
  
          system.terminate()
        },
        (11 * eventsPerActor).seconds)
      system.terminate() // just in case
    }
  
    def testPersistentActor(persistenceId: String): Props = Props(new TestPersistentActor(persistenceId))
  
    class TestPersistentActor(override val persistenceId: String) extends PersistentActor with ActorLogging {
      override def receiveRecover = PartialFunction.empty // no events to replay
  
      override def receiveCommand = {
        log.info("Recovered test persistent actor {}", persistenceId);
        {
          case s: String =>
            val replyTo = sender()
            persist(s) { _ =>
              replyTo ! Done
            }
        }
      }
    }
  }

[johanandren]

I think the idea has merit. Did you try running bench/test against actual database, with R2DBC for example?

[leviramsey]

running bench/test against actual database, with R2DBC for example?

Reproducing high-variance latency against a non-pathological backend would be difficult (short of introducing a pathological serializer which does some random Thread.sleeps, which would have other implications around dispatcher starvation), but that would be useful to demonstrate that this at least doesn't demonstrate a regression, so a good idea there.

[johanandren]

Sleeping inside a serializer is not the right simulation.

I think the interesting part would be delays from the outside, Toxiproxy with latency toxic and jitter, in front of a real DB to introduce latency for example.

An alternative could be something like a wrapper journal that delegates to the in-mem plugin and delays the completed future with the scheduler for ~1% of writes (say 100-500ms extra).

I think it would also be better with p50 and p99 percentiles persist round-trip latency with 1 vs 8 or 16 resequencers, since the claim here is that we improve latency. Total time isn't the right proof.

We'd also want the same runs with zero delay introduced, to confirm a non default value does not cost throughput when the store is healthy.

One aspect of this that I didn't think of initially is that for a struggling/overloaded database this multiple resequencers may actually be worse since the current head-of-line blocking kind of acts as accidental backpressure.

[leviramsey]

"accidental backpressure" is called out in the PR description ("To the extent that an occasional outlier is a symptom of an overloaded persistence backend, the resequencer introduces a small amount of backpressure on persists by delaying the next persist for an actor waiting for the head-of-line to unblock") and in the reference.conf comment ("Increasing ... removes a source of backpressure for a persistence backend where overload manifests as increased variance.")

With a slight tweak to make the resequencer specifics internal to the AsyncWriteJournal's receive (8302516), R2DBC runs the bench with non-default resequencer count.

The benchmark plugin effectively does a more severe version of "something like a wrapper journal that delegates to the in-mem plugin and delays the completed future with the scheduler" (potentially delaying for more than 10 seconds and imposing >500ms delay on more than 1% of persists), but it can be tweaked into that form. Percentiles on the persist asks can be added, too.

[leviramsey]

Using in-mem with a configurable percentage chance of delaying by 100-500ms (100 total actors, 100,100 total events; 4 core/8 thread x86 processor), times in ms

(ask latency / persist (measuring only within `asyncWriteMessages`) latency)    
  R   D   mean        stdev       p50         p90         p99         p999    
  1   0   1.3/1e-4    1.1/1e-4    1.2/1e-4    1.9/1e-4    4.8/1e-4    5.0/1e-4      
  1   1   154/3.9     165/36.1    115/1e-4    361/0.5     388/154     422/465    
  1   2   240/8.4     183/54.9    253/1e-4    436/0.2     454/355     455/489    
      
  2   0   0.7/1e-4    0.8/1e-4    0.3/1e-4    0.9/1e-4    3.6/1e-4    4.3/1e-4    
  2   1   130/5.6     185/44.8    0.4/0.1     460/0.2     499/208     504/507    
  2   2   158/5.9     176/44.6    129/3e-2    367/3e-2    379/324     380/480    
      
  4   0    1.4/1e-4   1.5/1e-4    1.0/1e-4    1.6/1e-4    6.7/1e-4    7.0/1e-4    
  4   1   68.5/3.4    139/34.6    0.3/5e-2    312/0.3     369/182     370/449    
  4   2    161/7.4    199/53.5    4.3/7e-2    451/0.5     452/362     455/493    
      
  8   0    1.2/1e-4   0.9/1e-4    0.9/1e-4    1.5/1e-4    4.9/1e-4    4.9/1e-4    
  8   1   50.2/3.9    128/37.8    0.3/1e-4    293/6e-2    428/177     504/496    
  8   2   82.9/6.1    155/46.6    1.1/1e-4    333/0.6     482/326     482/513    
      
  16  0    1.3/1e-4   1.0/1e-4     1.0/1e-4   2.0/1e-4    4.8/1e-4    4.9/1e-4    
  16  1   29.2/4.1    101/39.4    1e-4/3e-2   6.7/3e-2    461/195     541/499    
  16  2   64.0/7.2    142/51.9     1.1/1e-4   342/0.4     463/360     518/492    

Percentiles are a windowless estimate... the persist latency is only on the first 100k persists (~99.9% of the total), so the cases where ask latency is less than persist latency at a given percentile plausibly are due to the last asks completing faster (e.g. due to less delay or less contention)

[johanandren]

Can you follow up with:

• use HdrHistogram instead of the streaming estimator, with both latencies measured over the same writes
• An open-loop run with a fixed arrival rate per actor and rare delays. This matches the scenario in the PR description.
• Three or more runs per configuration, or a warmup phase followed by a measured phase.
• Throughput for the rows without delay, and the R2DBC numbers.
• Share the sources for the second benchmark in the PR or elsewhere so we can verify it is a correct bench

[leviramsey]

 (ask latency/all-persist latency/delayed-persist latency): 20k warmup (not tracked), 50k persists randomly distributed over 100 actors @ 1 ask per 10ms (in-mem plugin)
  
  R   D%  mean            stdev               P50             P90             P99                 P999                Nr delayed
   1  0   103us/1us/NM     135us/3us/NM       0ms/0ms/NM        0ms/0ms/NM       0ms/0ms/NM          2ms/0ms/NM          0
   1  1    57ms/3ms/317ms 113ms/34ms/114ms    0ms/0ms/310ms   240ms/0ms/479ms 450ms/129ms/519ms   580ms/479ms/520ms    506
   1  2   107ms/6ms/320ms 147ms/48ms/117ms    0ms/0ms/310ms   340ms/0ms/480ms 519ms/310ms/519ms   761ms/500ms/520ms   1002
   
   2  0   109us/1us/NM    113us/2us/NM        0ms/0ms/NM        0ms/0ms/NM      0ms/0ms/NM          1ms/0ms/NM           0
   2  1    32ms/3ms/324ms  89ms/35ms/117ms    0ms/0ms/320ms   139ms/0ms/481ms 420ms/130ms/520ms   511ms/489ms/520ms    514
   2  2    64ms/7ms/328ms 122ms/49ms/117ms    0ms/0ms/329ms   269ms/0ms/489ms 470ms/330ms/520ms   639ms/509ms/520ms   1004
   
   4  0   124us/2us/NM    87us/3us/NM         0ms/0ms/NM        0ms/0ms/NM       0ms/0ms/NM          0ms/0ms/NM          0     
   4  1    18ms/3ms/322ms 69ms/35ms/114ms     0ms/0ms/310ms     0ms/0ms/479ms 380ms/139ms/519ms   499ms/479ms/520ms    520  
   4  2    36ms/7ms/329ms 97ms/49ms/117ms     0ms/0ms/330ms   159ms/0ms/489ms 441ms/330ms/519ms   550ms/509ms/520ms   1004
   
   8  0   126us/1us/NM    83us/3us/NM         0ms/0ms/NM       0ms/0ms/NM        0ms/0ms/NM          0ms/0ms/NM          0
   8  1    11ms/3ms/321ms 55ms/34ms/116ms     0ms/0ms/319ms    0ms/0ms/480ms  330ms/129ms/520ms   499ms/480ms/520ms    514
   8  2    24ms/7ms/332ms 82ms/51ms/117ms     0ms/0ms/339ms   30ms/0ms/489ms  429ms/350ms/520ms   520ms/510ms/520ms   1082
   
  16  0   112us/1us/NM    99us/3us/NM         0ms/0ms/NM      0ms/0ms/NM        0ms/0ms/NM           0ms/0ms/NM          0
  16  1     8ms/3ms/326ms 47ms/34ms/118ms     0ms/0ms/329ms   0ms/0ms/489ms   299ms/0ms/520ms     489ms/489ms/520ms    491
  16  2    15ms/7ms/324ms 67ms/49ms/118ms     0ms/0ms/329ms   0ms/0ms/489ms   390ms/330ms/520ms   510ms/509ms/520ms   1039

bench:

 class DelayingInMemPlugin(config: Config) extends AsyncWriteJournal with AsyncRecovery {                                               
    // Instantaneous recovery, a miracle                                                                                                 
    override def asyncReadHighestSequenceNr(persistenceId: String, fromSequenceNr: Long): Future[Long] = zeroFuture                      
                                                                                                                                         
    override def asyncReplayMessages(persistenceId: String, fromSequenceNr: Long, toSequenceNr: Long, max: Long)(                        
        recoveryCallback: PersistentRepr => Unit): Future[Unit] =                                                                        
      Future.unit                                                                                                                        
                                                                                                                                         
    override def asyncWriteMessages(messages: Seq[AtomicWrite]): Future[Seq[Try[Unit]]] = {                                              
      if (schedulerGranularityMillis < 1) {                                                                                              
        schedulerGranularityMillis =                                                                                                     
          context.system.settings.config.getDuration("akka.scheduler.tick-duration").toMillis.min(Int.MaxValue / 2).toInt                
      }                                                                                                                                  
                                                                                                                                         
      val rand = ThreadLocalRandom.current()                                                                                             
                                                                                                                                         
      val delayMillis = {                                                                                                                
        val shouldDelay = rand.nextInt(100) < delayChance                                                                                
        if (shouldDelay) {                                                                                                               
          100 + rand.nextInt(400)                                                                                                        
        } else {                                                                                                                         
          0                                                                                                                              
        }                                                                                                                                
      }                                                                                                                                  
                                                                                                                                         
      val recordMeta =                                                                                                                   
        if (trackers ne initialTrackers)                                                                                                 
          (true, System.nanoTime(), trackers._1, trackers._2)                                                                            
        else initialMeta                                                                                                                 
                                                                                                                                         
      val fut = Future.successful(messages.map(_ => successUnit))                                                                                                                                          
                                                                                                                                                                                                           
      val ret =                                                                                                                                                                                            
        if (delayMillis == 0) fut                                                                                                                                                                          
        // asking for a shorter delay than what the scheduler can deliver will have an expected delay of approximately the requested                                                                       
        // delay: if requesting 10% of the granularity, there's a 10% chance of delay by the granularity and a 90% chance of zero delay  
        else if (delayMillis < schedulerGranularityMillis && rand.nextInt(schedulerGranularityMillis) > delayMillis) fut                 
        else after(delayMillis.millis)(fut)(context.system)                                                                              
                                                                                                                                         
      ret.andThen { _ =>                                                                                                                 
        if (recordMeta._1) {                                                                                                             
         val end = System.nanoTime()
          val (_, start, allPersists, delayedPersists) = recordMeta
          val duration = (end - start) / 1000
  
          allPersists ! duration
          if (delayMillis > 0) delayedPersists ! duration
        }
      }(ExecutionContext.parasitic)
    }
  
    override def asyncDeleteMessagesTo(persistenceId: String, toSequenceNr: Long): Future[Unit] = Future.unit
  
    override def receivePluginInternal: Receive = {
      case akka.Benchmark.JournalTrackers(allPersists, delayedPersists) =>
        trackers = allPersists -> delayedPersists
        sender() ! Done
    }
  
    val zeroFuture = Future.successful(0L)
    val successUnit = Success(())
    val delayChance = config.getInt("percent-chance-of-delay").min(100).max(0)
    val initialTrackers = ActorRef.noSender -> ActorRef.noSender
    val initialMeta = (false, 0L, ActorRef.noSender, ActorRef.noSender)
  
    var trackers: (ActorRef, ActorRef) = initialTrackers
    var schedulerGranularityMillis: Int = -1
  }
  
  package akka {
  
    import java.util.concurrent.atomic.AtomicInteger
    object Benchmark {
      def config(resequencers: Int, delayChance: Int) = ConfigFactory.parseString(s"""
      akka.persistence.journal.plugin = delaying-in-mem
  
      delaying-in-mem {
        class = "DelayingInMemPlugin"
        num-resequencers = $resequencers
        circuit-breaker.call-timeout = 11s
        percent-chance-of-delay = $delayChance
      }
      """).withFallback(ConfigFactory.load())
  
      def main(args: Array[String]): Unit = {
        val resequencers = if (args.length > 0) args(0).toInt else 1
        val delayPct = if (args.length > 1) args(1).toInt else 1
        val totalEvents = if (args.length > 2) args(2).toInt else 100000
        val numActors = if (args.length > 3) args(3).toInt.min(totalEvents) else 1000
  
        val WarmupEvents = 20000
  
        implicit val system = ActorSystem("Resequencer-Benchmark", config(resequencers, delayPct))
        import system.dispatcher
  
        val actors = (1 to numActors).iterator
          .map(i => system.actorOf(testPersistentActor(s"test-persistent-${UUID.randomUUID()}"), s"test-persistent-$i"))
          .toArray
  
        val askTracker = system.actorOf(trackerProps("ask-tracker"), "tracker")
        val allPersistTracker = system.actorOf(trackerProps("all-persists"), "all-persist-tracker")
        val delayedPersistTracker = system.actorOf(trackerProps("delayed-persists"), "delayed-persist-tracker")
  
        implicit val timeout: Timeout = 20.seconds
  
        system.log.info("Using {} resequencers", system.settings.config.getInt("delaying-in-mem.num-resequencers"))
  
        val warmup =
          Source
            .repeat(())
            .take(WarmupEvents.toLong)
            .zipWithIndex
            .mapAsync(numActors) {
              case (_, n) =>
                val actor = actors(n.toInt % numActors)
                actor ? "persist"
            }
            .runWith(Sink.ignore)
            .flatMap { _ =>
              // inject the trackers into the journal
              val journal = Persistence(system).journalFor("")
  
              journal ? JournalTrackers(allPersistTracker, delayedPersistTracker)
            }
  
        val stream = warmup.flatMap { _ =>
          system.log.info("Finished warmup")
  
          val count = new AtomicInteger()
  
          Source
            .repeat(())
            .take(totalEvents.toLong)
            .throttle(1, 10.millis)
            .mapAsyncUnordered(1000) { _ =>
              val rand = ThreadLocalRandom.current()
              val actor = actors(rand.nextInt(numActors))
  
              val start = System.nanoTime()
              (actor ? "persist").andThen { _ =>
                val finish = System.nanoTime()
  
                val n = count.incrementAndGet()
  
                if ((n % 1000) == 0) {
                  system.log.info("Successful ask #{}", n)
                }
  
                askTracker ! ((finish - start) / 1000)
              }(ExecutionContext.parasitic)
            }
            .runWith(Sink.ignore)
            .andThen { _ =>
              system.log.info("Finished run")
            }
        }
  
        Await.result(stream.flatMap { _ =>
          (askTracker ? "Log").flatMap { _ =>
            (allPersistTracker ? "Log").flatMap { _ =>
              (delayedPersistTracker ? "Log")
            }
          }
        }, (totalEvents / 50).seconds)
  
        system.terminate()
      }
  
      case class JournalTrackers(allPersists: ActorRef, delayedPersists: ActorRef)
  
      def testPersistentActor(persistenceId: String): Props = Props(new TestPersistentActor(persistenceId))
  
      class TestPersistentActor(override val persistenceId: String) extends PersistentActor with ActorLogging {
        override def receiveRecover = PartialFunction.empty // no events to replay
  
        override def receiveCommand = {
          case s: String =>
            val replyTo = sender()
            persist(s) { _ =>
              replyTo ! Done
            }
        }
      }
  
      def trackerProps(id: String): Props = Props(new PercentileTracker(id))
  
      class PercentileTracker(id: String) extends Actor with ActorLogging {
        val histogram = new Histogram(3)
        val receive = {
          case micros: Long => histogram.recordValue(micros)
          case "Log" =>
            log.info("[{}] observed {} measurements", id, histogram.getTotalCount)
            log.info("[{}] mean={}, stdev={}", id, histogram.getMean / 1000, histogram.getStdDeviation / 1000)
            log.info(
              "[{}] p50={}, p90={}",
              id,
              histogram.getValueAtPercentile(50.0) / 1000,
              histogram.getValueAtPercentile(90.0) / 1000)
            log.info(
              "[{}] p99={}, p999={}",
              id,
              histogram.getValueAtPercentile(99.0) / 1000,
              histogram.getValueAtPercentile(99.9) / 1000)
            sender() ! Done
        }
      }
    }
  }

[johanandren]

The in-memory numbers show that the resequencer itself stops being the bottleneck. They cannot show what happens when a real backend is the bottleneck, which is the case where anyone would turn this on. Please run the same open-loop bench against akka-persistence-r2dbc with Postgres. Three reasons:

1. Where the queueing moves. With one resequencer, a slow persist holds back the next WriteMessages from roughly 50 actors. With 16 resequencers those actors issue their next write right away, so the connection pool sees up to 16 times more concurrent writes during an outlier. The in-memory plugin does no work per write, so it cannot tell us whether that extra concurrency just moves the wait from the resequencer into the pool queue. With a real pool of, say, 20 connections, the p99 may get worse, not better.

2. The healthy backend case. The zero-delay rows in the in-memory bench prove that the routing code is free. They do not prove that a non-default value is free when every write costs a few milliseconds of real I/O. We want to be able to say "set this to 8 and nothing gets worse on a healthy database". That needs a real database.

3. The backpressure concern. We agree the single resequencer acts as accidental backpressure. Whether removing it hurts an overloaded Postgres is an empirical question. A run with a latency toxic (Toxiproxy, 200 to 500 ms with jitter on 1 percent of connections, or a periodic latency spike) in front of Postgres is the closest thing to the outlier scenario in the PR description.

Concretely, for R = 1, 8 and 16:

• no toxic: throughput plus p50/p99/p999 ask latency
• latency toxic: the same
• pool size at the r2dbc default, and a note on the CPU and connection count during the run

The result decides the recommendation we put in the reference.conf comment. If p99 improves without hurting throughput or the healthy case, the PR is good to go and we can suggest a value. If the pool becomes the bottleneck, the comment should say so and the setting stays a niche tuning knob.

[leviramsey]

Open-loop with R2DBC

• connection pool of at most 20 connections (warmup established all of them; none were observed to close)

• postgres is on localhost; 4 core/8 thread Core i7-8565U (4.6 GHz max clockspeed)

• R2DBC pubsub was disabled (no subscribers)

• toxic injects delays into the TCP traffic, with approximately 1% of traffic directly delayed (head-of-line blocking in TCP may result in more queries being delayed) by 100-500ms (logs of slow DB calls by the R2DBC journal are consistent with this: max delay was ~500ms)

• non-toxic, 1 resequencer: throughput=100/sec, mean=3ms, p50=3ms, p90=3ms, p99=5ms, p999=10ms

• non-toxic, 2 resequencers: throughput=100/sec, mean=2ms, p50=1ms, p90=3ms, p99=5ms, p999=8ms

• non-toxic, 8 resequencers: throughput=100/sec, mean=3ms, p50=3ms, p90=3ms, p99=5ms, p999=10ms

• non-toxic, 16 resequencers: throughput=100/sec, mean=3ms, p50=2ms, p90=3ms, p99=5ms, p999=9ms

• toxic: 1 resequencer: throughput=99/sec, mean=62ms, p50=10ms, p90=182ms, p99=271ms, p999=411ms

• toxic: 2 resequencers: throughput=100/sec, mean=38ms, p50=3ms, p90=150ms, p99=252ms, p999=373ms

• toxic: 8 resequencers: throughput=100/sec, mean=18ms, p50=3ms, p90=50ms, p99=241ms, p999=300ms

• toxic: 16 resequencers: throughput=100/sec, mean=14ms, p50=3ms, p90=4ms, p99=232ms, p999=272ms

One open question is if it's worth allowing num-resequencers to be zero (disable resequencing). Even with 16, there can still be head-of-line blocking if the max journal latency becomes greater than (num-resequencers times 1s/throughput: i.e. at 16 resequencers and 100 persists/sec, a 160ms asyncWriteMessages call is likely to result in at least one other persist being delayed).

[johanandren]

Thanks for the R2DBC numbers btw, looks good.

[patriknw]

Looking good. 🤖 review:

• [blocker] akka-persistence/src/main/scala/akka/persistence/journal/AsyncWriteJournal.scala:56
  — num-resequencers of 0 or a negative value is never validated. (1 to n) for n <= 0
  yields an empty Vector at construction, so nothing fails at startup; the first write then
  evaluates unsignedHash % 0 at line 64, throws ArithmeticException: / by zero inside
  receive, the journal restarts, and the persistent actor waits forever for a reply that
  never comes. This is precisely the failure mode the requirements rule out. Needs
  require(numResequencers >= 1, ...) at journal construction, where the operator sees it.
  (reference.conf:155 already claims "Default is 1, which is the minimum"; the code does
  not enforce it.)

• [blocker] akka-persistence/src/test/scala/akka/persistence/PersistentActorJournalProtocolSpec.scala:96
  — the new test does not exercise the change. The puppet journal is
  class JournalPuppet extends Actor, a plain forwarding actor, not an
  AsyncWriteJournal: it never reads num-resequencers, never creates a Resequencer, and
  confirm() (lines 130-137) sends WriteMessagesSuccessful/WriteMessageSuccess directly
  to the persistent actor, bypassing any resequencer. The OneResequencer/TwoResequencers
  parameterisation and the new case at line 268 are therefore both inert. Verified: the new
  test passes with -Dpuppet.num-resequencers=1. Nothing on this branch would fail if every
  actor were sent to one group, or if a write were answered from a group other than the one
  it reserved from.
  Smallest discriminating test: use SteppingInmemJournal
  (akka-persistence/src/test/scala/akka/persistence/journal/SteppingInmemJournal.scala, a
  real InmemJournal whose writes complete on demand) with num-resequencers = 2; start
  persistent actors until two land in different groups
  ((ref.hashCode & 0x7FFFFFFF) % 2 differs), let both persist, release only the second
  actor's write and assert its persist callback runs while the first write is still
  outstanding. That times out on main and passes on the branch. Pair it with the same
  scenario at num-resequencers = 1 asserting the reverse, to pin the default.

• [blocker] akka-persistence/src/test/scala/akka/persistence/PersistentActorJournalProtocolSpec.scala:284
  — the new test is flaky. Rejected candidates are stopped with PoisonPill, but their
  PostStop(test-N) monitor message (line 48) is never consumed, so it races with the next
  startActor's expectMsg(PreStart/RecoveryCompleted/Done) and with the Done assertions
  later in the test. Observed 5 failures in 11 runs, all
  assertion failed: expected <X>, found PostStop(test-8). Either expectMsg(PostStop(pid))
  after each PoisonPill, or pick the second actor without starting/stopping candidates —
  the group is a function of ref.hashCode, so a plain system.actorOf loop that stops
  nothing and only keeps the matching ref would do.

• [should-fix] akka-persistence/src/test/scala/akka/persistence/PersistentActorJournalProtocolSpec.scala:105,265-266
  — class PersistentActorJournalProtocolSpec(n: Int) stays concrete while losing its
  no-arg constructor, which is a test-discovery hazard; it should be abstract. The two
  runnable suites are named OneResequencer and TwoResequencers, dropping the Spec
  suffix every other suite in the repo uses; prefer
  PersistentActorJournalProtocolOneResequencerSpec / ...TwoResequencersSpec.

• [nit] akka-persistence/src/main/resources/reference.conf:154-157 — double spaces
  after full stops (the rest of the file uses one). "assigned to a resequencer ... by hash"
  is imprecise: the hash is of the persistent actor's ActorRef (its incarnation uid), so
  an actor that is stopped and recreated with the same persistence id can land in a
  different group — worth saying, since a reader will assume persistence id. The
  backpressure sentence does follow from the mechanism and is correct.

• [nit] akka-persistence/src/main/resources/reference.conf:158 — num-resequencers
  names an internal class (Resequencer) in a permanent public configuration key. A user
  tuning a journal has no way to know what a resequencer is; something like
  write-reply-ordering-groups describes the observable behaviour.

• [nit] akka-persistence/src/main/scala/akka/persistence/journal/AsyncWriteJournal.scala:362
  — class ResequencerHandle can be final.

[johanandren]

Blocking

• AsyncWriteJournal.scala:52: the validation rejects valid configuration. It checks valueType != NUMBER, but values coming from -D system properties or ${?ENV_VAR} substitutions are strings, which getInt parses fine. With write-reply-ordering-groups = "4" the journal fails at startup with write-reply-ordering-groups must be a positive integer, and persistent actors then hang until recovery timeout.
• AsyncWriteJournal.scala:65: 0 and negative values are silently clamped to 1 via .max(1). As requested earlier these should fail journal startup. Verified locally that 0 and -3 start without any error.
• Both are solved by dropping the block at lines 49-56 (and the ConfigValueType import) and validating inside the receiveWriteJournal initializer instead. That initializer is an eager final val, so it already runs at construction:

  val numResequencers = config.getInt("write-reply-ordering-groups")
  require(numResequencers >= 1, s"write-reply-ordering-groups must be >= 1, was [$numResequencers]")

  Please add a test that 0 fails journal startup.

Should fix

• ControlledInMemJournal.scala:96: asyncWriteMessages returns promise.future instead of superPromise.future. The write is acknowledged before the event is stored in memory, and a failure from the real write (for example serialization) is dropped. superPromise is otherwise unused.
• JournalReplyOrderingSpec: test cases are not isolated. All cases share the same actors and journal, so one failing case leaves promises uncompleted and the following cases fail with unrelated receiveN timeouts. In my mutation runs one real failure turned into four. Complete any outstanding promises in a cleanup step, or use fresh actors per case.
• JournalReplyOrderingSpec.scala:198-202: drop the 4 and 5 group variants. They account for 840 of the 872 test cases and exercise no behaviour beyond 1, 2 and 3 groups.
• Coverage with more than one group. Every write in the spec carries exactly one event. Could you add a case with persistAll or defer (counter advancing by more than 2) and one with a failed write, for a group count above 1? CI only runs the existing suites with one group.
• reference.conf:154-158: the comment does not say what the setting is. Suggestion, also wrapped like the neighbouring comments:

  # Number of groups used for ordering the write replies from the journal.
  # Replies within a group are delivered in the order the writes were
  # received, so one slow write delays the replies of all later writes in
  # the same group. A persistent actor is assigned to a group by the uid
  # of its ActorRef.
  # Increasing this reduces such head-of-line blocking when write latency
  # varies a lot, but it also removes a source of backpressure towards an
  # overloaded database. Must be at least 1.
  

  Given the R2DBC numbers, should we also suggest a value, for example 8 or 16?
• Documentation. The setting is not mentioned in the paradox docs. Should it be, or is reference.conf enough for a tuning knob like this?

Nits

• AsyncWriteJournal.scala:66-68: Vector.fill(numResequencers)(...) instead of (1 to n).iterator.map(_ => ...).toVector.
• AsyncWriteJournal.scala:73-74: Math.floorMod(actor.hashCode, numResequencers) instead of the sign-bit mask and %.
• AsyncWriteJournal.scala:79-83: capture handle.resequencer in a local val before the callback, so the closure holds no mutable object and the two-line safety comment is not needed.
• JournalReplyOrderingSpec.scala:78: the spec duplicates the routing formula. A private[persistence] helper in the AsyncWriteJournal companion used by both would keep them in sync.
• There is no upper bound on the setting. Each group is an actor created at journal startup, so a typo like 100000 goes unnoticed. Worth a cap or a logged warning?
• JournalReplyOrderingSpec.scala:198-202: OneResequencerSpec etc. are generic names, and the setting is no longer called resequencers. Something like JournalReplyOrderingOneGroupSpec.
• JournalReplyOrderingSpec.scala:110: the test name starts with "must" inside a must block, reports read "must must properly order".
• JournalReplyOrderingSpec.scala:88: Random.nextInt makes a failure non-reproducible. Use a fixed group, or log the chosen one.
• JournalReplyOrderingSpec.scala:152: the assertion "futures in other groups should not be completed" cannot fail. It runs immediately after completing the promise, and mostly checks asks whose own write has not completed yet.
• JournalReplyOrderingSpec.scala:145-189: the fold is hard to follow, with special cases for the group of two and for reversed completion. A model would be simpler: after each step an ask is expected to be complete if its write completed and no earlier write in its group is still open.
• JournalReplyOrderingSpec.scala:136-141: reversedCompletion is ordering.indexOf(first) > ordering.indexOf(second).
• JournalReplyOrderingSpec.scala:69: journalInstanceId is read back from config but is always specName.
• JournalReplyOrderingSpec.scala:34-57: persistentActorCounter, persistentActorProps and oneSuccessUnit can be private.
• ControlledInMemJournal.scala: the file name says InMem, the class says Inmem.
• ControlledInMemJournal.scala:65-66: persistProbe and instanceId can be private.
• ControlledInMemJournal.scala:81,95: self instead of context.self, the second one is called from a future callback.
• New files: scaladoc starts on the /** line. The rest of the code base puts the text on the following line with *.

[leviramsey]

As far as recommending a value, it's going to vary based on the write rate and tail latency.

[johanandren]

Almost there. Some feedback still

Should fix

• ControlledInmemJournal.scala:29-33: the scaladoc promises rejections that cannot happen. The contents of the completed sequence are discarded. Only the length is checked, then the result of the real in-memory write is used. A Failure element never gives WriteMessageRejected. Either pass the results from the test through, or change the doc.
• LoadPluginSpec.scala:70-80: the actor system is not shut down reliably. terminate() is not awaited, and it is skipped when the assertion fails. Use TestKit.shutdownActorSystem(sys) in a finally. The test name also still says "resequencers".
• AsyncWriteJournal.scala:56: include the value in the require message, for example write-reply-ordering-groups must be >= 1, was [$numResequencers].
• JournalReplyOrderingSpec.scala:89-91: "cull" is never sent. Remove it.
• JournalReplyOrderingSpec.scala:415: the respawn message is consumed only at the end of the "first fails" cases. If such a case fails earlier, the message stays in the probe and the drain in afterEach miscounts for the next case.
• Open questions from the last round: should the setting be mentioned in the Paradox docs, and should there be an upper bound or a logged warning for very large values? Each group is an actor created at journal startup.

Simplifications in JournalReplyOrderingSpec

The spec is large compared to the change it covers. These would make it easier to maintain:

• Lines 317-416: the two test bodies are near copies. One helper that takes the pid to fail as an Option would halve the file.
• Lines 208-230: the waiting fold over Option[Boolean] can be

  attempts.dropWhile(_.promise.isCompleted).drop(1).exists(_.promise.isCompleted)

  which also removes SomeTrue and SomeFalse.
• Lines 204-207: a small case class instead of the (String, Int, Promise[...]) tuples. The _1 and _3 accesses hide what is compared.
• Lines 232-259: success and failure differ only in how the promise is completed.
• Lines 98-108: the var promise null handling in the persistAll handler can be p.trySuccess(Done).
• Lines 43-70: fresh actors per test case, created in a loop until the hash fits the wanted group, would remove the backoff supervisor, the self-stop in preStart, the spawn probe and the retry in resolvePid.

Nits

• reference.conf:155-160: double spaces after full stops. The rest of the file uses one.
• AsyncWriteJournal.scala:60-63: braces around a single expression in the else branch.
• AsyncWriteJournal.scala:344-346: ResequencerHandle can be private, hashForResequencing can be private[journal]. A helper that returns the group index for an actor and a group count would let the spec share the whole routing formula, not only the hash.
• ControlledInmemJournal.scala:85-87: persistProbe, instanceId and promises can be private.
• ControlledInmemJournal.scala:115-119: tryFailure on a promise that has already completed does nothing.
• ControlledInmemJournal.scala:144: self instead of context.self.
• ControlledInmemJournal.scala:46: the // comment above the scaladoc of getProbe can go.
• ControlledInmemJournal.scala:47,76: scaladoc text starts on the /** line. The rest of the code base puts it on the next line.
• JournalReplyOrderingSpec.scala:38,124: numResequencers should follow the name of the setting, for example groups.
• JournalReplyOrderingSpec.scala:146: the log text says "two resequencers", it is two persistent actors.
• JournalReplyOrderingSpec.scala:57-66: the preStart comment can be one line on why the actor stops itself.
• JournalReplyOrderingSpec.scala:49-50: 10.milli next to 10.millis.
• JournalReplyOrderingSpec.scala:130: the require duplicates the validation in the journal.
• JournalReplyOrderingSpec.scala:393: asks(pid).failed.futureValue.getMessage should include(...) is the usual form.
• JournalReplyOrderingSpec.scala:118-121,278-279: oneSuccessUnit, twoSuccessUnit, BackingOff, SomeTrue and SomeFalse can be private.
• JournalReplyOrderingSpec.scala:421-422: the journal instance ids "one" and "two" are keys in a JVM-global registry. Something longer, like the spec name, avoids a collision with another spec.
• Both new files: imports are not in project order.

[leviramsey]

• Open questions from the last round: should the setting be mentioned in the Paradox docs, and should there be an upper bound or a logged warning for very large values? Each group is an actor created at journal startup.

I've added a configurable limit defaulting to 10k (which would be less than 10 MiB of heap in typical openjdk-derived JVMs). If one really needs more, they then have to override both the regular and limit configurations.

reference.conf ends up incorporated into the paradox docs. It probably can/should be called out in the plugin paradox docs (especially since I'd expect that the usefulness of this varies by plugin: backends where events end up in different places like Cassandra, Dynamo, or R2DBC with database sharding are the ones where one would expect to see a big difference between the tail and the norm (and the throughput for the HOLB to have an impact).

[johanandren]

LGTM, but let's wait for a second Ok from @patriknw before we merge.

[patriknw]

LGTM

One remark from 🤖 about something that I think was also discussed in previous reviews.

• [should-fix] akka-persistence/src/main/scala/akka/persistence/journal/AsyncWriteJournal.scala:344: max-write-reply-ordering-groups at 0 or below is raised to 1 by .max(1) instead of being refused. reference.conf:169 says that it must be at least 1. With a limit of 0 and one group, the journal starts without an error. With a limit of 0 and write-reply-ordering-groups = 4, the journal fails with must have no more than [1] write-reply-ordering-groups, was [4]. That message names neither the limit setting nor its configured value (line 348). Refuse a limit below 1 the same way as the group count, and name max-write-reply-ordering-groups in both messages. (by reading)

[leviramsey]

OK, removed that max.