From 80481f756d71bd58b4bda627758dcd774e9d5dcd Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Fri, 18 Sep 2026 14:45:54 +0000
Subject: [PATCH] [#986] Give back the changes a replay thread the pool stopped had parked (#988)

---
 opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java |  544 +++++++++++++++++++++++++++++++++++++++++++++++++++++-
 1 files changed, 534 insertions(+), 10 deletions(-)

diff --git a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java
index 123ace3..325212b 100644
--- a/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java
+++ b/opendj-server-legacy/src/test/java/org/opends/server/replication/UpdateOperationTest.java
@@ -95,6 +95,7 @@
 import org.opends.server.types.Operation;
 import org.opends.server.types.OperationType;
 import org.opends.server.types.RawModification;
+import org.opends.server.util.StaticUtils;
 import org.opends.server.util.TestTimer;
 import org.opends.server.util.TestTimer.CallableVoid;
 import org.opends.server.util.TimeThread;
@@ -3583,6 +3584,508 @@
     }
   }
 
+  /**
+   * Test case for [Issue 954]: a change parked as waiting for another one is given back
+   * when the replay which parked it is unwound, and the session is restarted for it.
+   * <p>
+   * A change which waits for another one is parked and stays owned by the replay thread
+   * which parked it, while that thread goes on to the changes which follow: it is handed
+   * out again by {@code getNextUpdate()}, which every replay loop of this domain runs once
+   * it is done, so it is replayed by whichever thread clears the change it was waiting for.
+   * A replay which is unwound leaves the thread which parked it without that road - it
+   * takes the next delivery off the shared queue instead, and never comes back to the
+   * change it parked - and every redelivery of a change a replay thread owns is refused as
+   * a duplicate. On a domain which then goes quiet that change is where this replica's
+   * ServerState, and every change behind it from every master, stops.
+   * <p>
+   * The replay which is unwound here is one which applied its change: the change it was
+   * replaying is committed and owns nothing anymore by the time the give-back on the way
+   * out of {@code replay()} runs, so the restart that give-back asks for is the one thing
+   * which has the parked change delivered again - a replay which failed would have asked
+   * for the same restart on the road of its own change. The parked change travels the
+   * replication server, and nothing but a new session brings it back.
+   */
+  @Test
+  public void aChangeParkedByAnUnwoundReplayIsDeliveredAgain() throws Exception
+  {
+    testSetUp("aChangeParkedByAnUnwoundReplayIsDeliveredAgain");
+    logger.error(LocalizableMessage.raw(
+        "Starting replication test : aChangeParkedByAnUnwoundReplayIsDeliveredAgain"));
+
+    final DN waitedOn = addEntryForChange("user.954.1");
+    final String waitedOnUUID = getEntry(waitedOn, 1, true).parseAttribute("entryuuid").asString();
+    final DN other = addEntryForChange("user.954.2");
+    final String otherUUID = getEntry(other, 1, true).parseAttribute("entryuuid").asString();
+
+    final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+    domain.resetUnreplayedChangeAlertThrottle();
+    final long inProgress = getMonitorAttrValue(baseDN, "changes-in-progress-size");
+    assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
+        "no change of this domain is waiting for another one when this test starts");
+
+    // The replica the changes come from: one of its own, so that the CSNs of this test are
+    // never covered by the ServerState another one left behind.
+    final int serverId = 30;
+    final CSNGenerator gen = new CSNGenerator(serverId, TimeThread.getTime());
+    final CSN failing = gen.newCSN();
+    final CSN parked = gen.newCSN();
+    final CSN unwound = gen.newCSN();
+    final List<Modification> failingMods = generatemods("description", "the replay of this change fails");
+    final String parkedDescription = "the change which was parked as a dependency";
+    final List<Modification> parkedMods = generatemods("description", parkedDescription);
+    final List<Modification> unwoundMods =
+        generatemods("description", "the replay of this change is unwound once it is applied");
+
+    /*
+     * The change whose replay fails is the barrier the parked change waits behind, so its
+     * budget must not be spent while this test is setting up: it is shortened once the
+     * change which was parked is back, and put back in the finally below.
+     */
+    setReplayGiveUpDelay("unlimited");
+    try
+    {
+      /*
+       * One replay thread, so that the change which is parked and the replay which is
+       * unwound after it are the same thread's: a parked change is left owned by the thread
+       * which parked it, and this is about a thread which does not come back to it.
+       */
+      setNumUpdateReplayThreads(1);
+      try
+      {
+        /*
+         * The change which is parked is published to the replication server rather than
+         * handed to the domain: a change which travelled a session is one the replication
+         * server sends again over the next session of this domain, and over nothing else -
+         * which is what the restart the give-back asks for has to be pinned against.
+         */
+        final ReplicationBroker broker =
+            openReplicationSession(baseDN, serverId, 100, replServerPort, 1000);
+        try
+        {
+          /*
+           * A change whose replay failed stays listed and uncommitted - it is what holds this
+           * domain's ServerState back - and stays among the changes the newer ones are checked
+           * against, so a change which follows it on the same entry has to wait for it. It has
+           * to be a change which is asked for again rather than stepped over: one whose
+           * operation is built and then refused, since #928 has a modify whose entry DN does
+           * not parse reported once and recorded as replayed.
+           */
+          deliverUntilMonitorReaches(domain, "changes-in-progress-size", inProgress + 1,
+              () -> new ModifyMsgWhoseOperationRefusesAControl(failing, waitedOn, failingMods, waitedOnUUID),
+              "the change whose replay fails must be listed as one which is not in the data");
+
+          // The change which is parked as waiting for it by the replay thread it was given to.
+          broker.publish(new ModifyMsg(parked, waitedOn, parkedMods, waitedOnUUID));
+          assertMonitorAttrValueEventually(baseDN, "dependent-changes-size", 1,
+              "a change which waits for one that is not in the data must be parked");
+
+          /*
+           * How many restarts in a row the session has been through, the barrier's among
+           * them: the restart the give-back asks for on the road out of a JVM which has run
+           * out of memory is owed no backoff, and a restart which waits the backoff out is
+           * the one road which moves this count. It can not move otherwise between here and
+           * the reading below: a replay which made it puts the count back to zero only once
+           * nothing is failing anymore, and the barrier keeps failing until it is given up.
+           */
+          final int restarts = domain.getConsecutiveSessionRestarts();
+
+          /*
+           * The replay which is unwound while that same thread still holds the parked change.
+           * It applies its change, and the ack of its delivery is where the JVM runs out of
+           * memory: that is the one throw from the ack which is not caught, past every catch
+           * the replay itself has, and it leaves replay() with the change this thread was
+           * replaying committed and owned by nobody. What the give-back on the way out finds
+           * to give back is the parked change alone, and the restart it asks for is the one
+           * road which has that change delivered again: the road a failed replay takes to
+           * ask for its own change is not run. An Error met replaying a change does not get
+           * here, and neither does any other throw from the ack: both are reported and the
+           * replay carries on to the road the change itself decided (issue #922).
+           *
+           * It is made on another entry, so that it is replayed rather than parked in its
+           * turn, and it is delivered until the parked change has been given back: a delivery
+           * of a change which is being replayed, or was applied, is refused as the duplicate
+           * it is.
+           */
+          deliverUntilMonitorReaches(domain, "dependent-changes-size", 0,
+              () -> new ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(unwound, other, unwoundMods, otherUUID),
+              "a change parked by a replay which was unwound must be given back");
+
+          /*
+           * The thread which met the error is gone - an OutOfMemoryError ends the replay
+           * thread which met it (issue #923) - and it was the whole pool, so the delivery the
+           * restarted session brings waits in the replay queue: the pool is brought back for
+           * it, at any number now that the thread it had to be the same as is over. The change
+           * is parked again by the thread which takes it, the barrier being still there: that
+           * it is parked at all is what says a new session delivered it.
+           */
+          setNumUpdateReplayThreads(2);
+          assertMonitorAttrValueEventually(baseDN, "dependent-changes-size", 1,
+              "the change which was given back must be delivered again by the session which"
+                  + " was restarted for it");
+          assertEquals(domain.getConsecutiveSessionRestarts(), restarts,
+              "a thread an OutOfMemoryError is ending must not wait the backoff out before"
+                  + " it restarts the session for the changes it gave back");
+
+          /*
+           * The barrier is lifted, which lets the ServerState past it and hands the parked
+           * change to the thread which cleared it.
+           */
+          giveUpOn(domain, failing, waitedOn, failingMods, waitedOnUUID);
+          checkEntryHasAttributeValue(waitedOn, "description", parkedDescription, 30,
+              "the change which was parked must be applied by the delivery which took it over");
+        }
+        finally
+        {
+          broker.stop();
+        }
+        assertEquals(resetNumUpdateReplayThreads(), 0,
+            "the number of replay threads could not be put back");
+      }
+      finally
+      {
+        /*
+         * Best-effort on the way out of a red, with no assertion to replace the failure
+         * which is being reported: on the normal road the number is put back, and checked,
+         * above.
+         */
+        resetNumUpdateReplayThreads();
+      }
+    }
+    finally
+    {
+      try
+      {
+        /*
+         * A red before the barrier was given up would leave it listed for the rest of this
+         * class: it never travelled the replication server, so no session restart brings a
+         * delivery which could give it up, and a change which is failing keeps every later
+         * session restart at its backoff and the ServerState behind it. The failure which
+         * is being reported is the one that matters, so this does not replace it.
+         */
+        if (!domain.getServerState().cover(failing))
+        {
+          giveUpOn(domain, failing, waitedOn, failingMods, waitedOnUUID);
+        }
+      }
+      catch (Throwable cleanupFailure)
+      {
+        logger.error(LocalizableMessage.raw(
+            "the barrier of aChangeParkedByAnUnwoundReplayIsDeliveredAgain could not be given"
+                + " up on the way out: %s", StaticUtils.stackTraceToSingleLineString(cleanupFailure)));
+      }
+      finally
+      {
+        resetReplayGiveUpDelay();
+      }
+    }
+  }
+
+  /**
+   * Gives up on the change no delivery can replay: the budget is shortened to nothing, so
+   * that the change is given up on as soon as one more delivery of it fails. Nothing sends
+   * that change again - it never travelled a session - so its deliveries are made here.
+   */
+  private void giveUpOn(final LDAPReplicationDomain domain, final CSN failing, final DN dn,
+      final List<Modification> mods, final String entryUUID) throws Exception
+  {
+    setReplayGiveUpDelay("0ms");
+    TestTimer timer = new TestTimer.Builder()
+      .maxSleep(120, SECONDS)
+      .sleepTimes(500, MILLISECONDS)
+      .toTimer();
+    timer.repeatUntilSuccess(new CallableVoid()
+    {
+      @Override
+      public void call() throws Exception
+      {
+        if (!domain.getServerState().cover(failing))
+        {
+          domain.processUpdate(new ModifyMsgWhoseOperationRefusesAControl(failing, dn, mods, entryUUID));
+        }
+        assertTrue(domain.getServerState().cover(failing),
+            "the change no delivery can replay must be given up on");
+      }
+    });
+  }
+
+  /**
+   * Test case for [Issue 986]: a change parked as waiting for another one is given back
+   * when the replay thread which parked it is stopped with the pool.
+   * <p>
+   * A parked change stays owned by the replay thread which parked it while that thread
+   * goes back to the pool and takes the changes which follow: {@code getNextUpdate()} is
+   * what hands it out again, to whichever replay thread clears the change it was waiting
+   * for. Changing the number of replay threads stops the whole pool and creates another
+   * one, so a thread which parked a change and went back to the queue is joined while it
+   * is idle, and it would end still recorded as the owner of that change - a thread which
+   * does not exist anymore, while every redelivery of a change a replay thread owns is
+   * refused as a duplicate. On a domain which then goes quiet that change is where this
+   * replica's ServerState, and every change behind it from every master, stops.
+   */
+  @Test
+  public void aChangeParkedByAThreadThePoolStoppedIsDeliveredAgain() throws Exception
+  {
+    testSetUp("aChangeParkedByAThreadThePoolStoppedIsDeliveredAgain");
+    logger.error(LocalizableMessage.raw(
+        "Starting replication test : aChangeParkedByAThreadThePoolStoppedIsDeliveredAgain"));
+
+    final DN waitedOn = addEntryForChange("user.986.1");
+    final String waitedOnUUID = getEntry(waitedOn, 1, true).parseAttribute("entryuuid").asString();
+    final DN other = addEntryForChange("user.986.2");
+    final String otherUUID = getEntry(other, 1, true).parseAttribute("entryuuid").asString();
+
+    final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
+    domain.resetUnreplayedChangeAlertThrottle();
+    final long inProgress = getMonitorAttrValue(baseDN, "changes-in-progress-size");
+    assertEquals(getMonitorAttrValue(baseDN, "dependent-changes-size"), 0,
+        "no change of this domain is waiting for another one when this test starts");
+
+    final CSNGenerator gen = new CSNGenerator(26, TimeThread.getTime());
+    final CSN failing = gen.newCSN();
+    final CSN parked = gen.newCSN();
+    final CSN applied = gen.newCSN();
+    final List<Modification> failingMods = generatemods("description", "the replay of this change fails");
+    final String parkedDescription = "the change which was parked by a thread the pool stopped";
+    final List<Modification> parkedMods = generatemods("description", parkedDescription);
+    final String appliedDescription = "the change which is applied before the pool is stopped";
+    final List<Modification> appliedMods = generatemods("description", appliedDescription);
+
+    /*
+     * One replay thread, so that the change which is parked is parked by the thread the
+     * pool then stops: a parked change is left owned by the thread which parked it, and
+     * this is about a thread which is not there anymore to be given it back.
+     */
+    setNumUpdateReplayThreads(1);
+    try
+    {
+      /*
+       * A change whose replay failed stays listed and uncommitted - it is what holds this
+       * domain's ServerState back - and stays among the changes the newer ones are checked
+       * against, so a change which follows it on the same entry has to wait for it.
+       */
+      deliverUntilMonitorReaches(domain, "changes-in-progress-size", inProgress + 1,
+          () -> new ModifyMsgWhoseOperationRefusesAControl(failing, waitedOn, failingMods, waitedOnUUID),
+          "the change whose replay fails must stay listed as one which is not in the data");
+
+      /*
+       * The change which is parked as waiting for it. The replay thread it was given to
+       * parks it and goes back to the pool: nothing is waiting for it there, so it is idle
+       * and it still owns the change it parked.
+       */
+      deliverUntilMonitorReaches(domain, "dependent-changes-size", 1,
+          () -> new ModifyMsg(parked, waitedOn, parkedMods, waitedOnUUID),
+          "a change which waits for one that is not in the data must be parked");
+
+      /*
+       * A change which is applied, on another entry so that it is replayed rather than
+       * parked in its turn. It is there for the count of the deliveries this session took
+       * off it, replayed-updates: a session which is started counts from zero, so that
+       * count going back to zero below is what says the session was restarted - and it is
+       * not left to the give-back alone to put the count above zero before the pool is
+       * stopped. The restart the change whose replay failed asked for is over by now: the
+       * thread which ran it is the one which parked the change above.
+       */
+      domain.processUpdate(new ModifyMsg(applied, other, appliedMods, otherUUID));
+      checkEntryHasAttributeValue(other, "description", appliedDescription, 30,
+          "the change made on the other entry must be applied");
+      // Counted once its ack is out, which is after the change is in the data.
+      final TestTimer counted = new TestTimer.Builder()
+        .maxSleep(30, SECONDS)
+        .sleepTimes(200, MILLISECONDS)
+        .toTimer();
+      counted.repeatUntilSuccess(new CallableVoid()
+      {
+        @Override
+        public void call() throws Exception
+        {
+          assertTrue(getMonitorAttrValue(baseDN, "replayed-updates") > 0,
+              "the session must have counted the deliveries it took off before the pool is stopped");
+        }
+      });
+
+      /*
+       * How many restarts in a row the session has been through, the barrier's among them:
+       * the restart the give-back of a stopped thread asks for is owed no backoff - what
+       * went away is a replay thread, not the backend - and a restart which waits the
+       * backoff out is the one road which moves this count. It can not move otherwise
+       * between here and the reading below: a replay which made it puts the count back to
+       * zero only once nothing is failing anymore, and the barrier keeps failing until it
+       * is given up, below. The restart the barrier asked for is over: the thread which
+       * ran it is the one which parked the change and applied the other, above.
+       */
+      final int restarts = domain.getConsecutiveSessionRestarts();
+
+      /*
+       * The pool is stopped and created again, the way an administrator changing the
+       * number of replay threads has it: the thread which parked the change is joined
+       * where it waits for the next delivery, and the change it owns is nobody's.
+       */
+      setNumUpdateReplayThreads(2);
+
+      assertMonitorAttrValueEventually(baseDN, "dependent-changes-size", 0,
+          "a change parked by a replay thread the pool stopped must be given back");
+      /*
+       * The give-back alone leaves the count where it was, plus the change it handed back:
+       * only a session which is started counts from zero, and the changes of this test
+       * never travelled a session, so nothing is delivered over the one which is started
+       * before the deliveries made below.
+       */
+      assertMonitorAttrValueEventually(baseDN, "replayed-updates", 0,
+          "the session must be restarted for the changes which were given back");
+      assertEquals(domain.getConsecutiveSessionRestarts(), restarts,
+          "the restart run for the changes a stopped thread gave back must not wait the"
+              + " backoff out: what went away is a replay thread, not the backend");
+
+      /*
+       * Nothing sends these changes again - they never travelled a session - so the
+       * deliveries which take over from the ones the stopped pool left behind are made
+       * here. The change no delivery can replay is given up on, which lets the ServerState
+       * past it, and the change which was parked behind it is applied.
+       */
+      setReplayGiveUpDelay(TEST_GIVE_UP_DELAY);
+      TestTimer timer = new TestTimer.Builder()
+        .maxSleep(120, SECONDS)
+        .sleepTimes(500, MILLISECONDS)
+        .toTimer();
+      timer.repeatUntilSuccess(new CallableVoid()
+      {
+        @Override
+        public void call() throws Exception
+        {
+          final ServerState state = domain.getServerState();
+          if (!state.cover(failing))
+          {
+            domain.processUpdate(
+                new ModifyMsgWhoseOperationRefusesAControl(failing, waitedOn, failingMods, waitedOnUUID));
+          }
+          if (!state.cover(parked))
+          {
+            domain.processUpdate(new ModifyMsg(parked, waitedOn, parkedMods, waitedOnUUID));
+          }
+          assertTrue(state.cover(parked),
+              "the change which was given back must be replayed by the delivery which takes it over");
+        }
+      });
+      checkEntryHasAttributeValue(waitedOn, "description", parkedDescription, 30,
+          "the change which was parked must be applied by the delivery which took it over");
+      assertEquals(resetNumUpdateReplayThreads(), 0,
+          "the number of replay threads could not be put back");
+    }
+    finally
+    {
+      try
+      {
+        /*
+         * A red before the barrier was given up would leave it listed for the rest of this
+         * class: it never travelled the replication server, so no session restart brings a
+         * delivery which could give it up, and a change which is failing keeps every later
+         * session restart at its backoff and the ServerState behind it. The failure which
+         * is being reported is the one that matters, so this does not replace it.
+         */
+        if (!domain.getServerState().cover(failing))
+        {
+          giveUpOn(domain, failing, waitedOn, failingMods, waitedOnUUID);
+        }
+      }
+      catch (Throwable cleanupFailure)
+      {
+        logger.error(LocalizableMessage.raw(
+            "the barrier of aChangeParkedByAThreadThePoolStoppedIsDeliveredAgain could not be"
+                + " given up on the way out: %s", StaticUtils.stackTraceToSingleLineString(cleanupFailure)));
+      }
+      finally
+      {
+        try
+        {
+          resetReplayGiveUpDelay();
+        }
+        finally
+        {
+          /*
+           * Best-effort on the way out of a red, with no assertion to replace the failure
+           * which is being reported: on the normal road the number is put back, and checked,
+           * above.
+           */
+          resetNumUpdateReplayThreads();
+        }
+      }
+    }
+  }
+
+  /**
+   * Delivers a change until a monitor attribute of the domain reaches the expected value.
+   * <p>
+   * A delivery is dropped rather than queued while the listener thread is down, which it
+   * is for as long as a recovery is restarting the session, so a change which has to reach
+   * a replay thread is delivered until it does. A delivery of a change a replay thread
+   * owns is refused as the duplicate it is, so the deliveries which follow the one that
+   * was taken cost nothing.
+   */
+  private void deliverUntilMonitorReaches(final LDAPReplicationDomain domain,
+      final String attributeName, final long expected,
+      final Supplier<? extends LDAPUpdateMsg> delivery, final String message) throws Exception
+  {
+    TestTimer timer = new TestTimer.Builder()
+      .maxSleep(20, SECONDS)
+      .sleepTimes(500, MILLISECONDS)
+      .toTimer();
+    timer.repeatUntilSuccess(new CallableVoid()
+    {
+      @Override
+      public void call() throws Exception
+      {
+        if (getMonitorAttrValue(baseDN, attributeName) != expected)
+        {
+          domain.processUpdate(delivery.get());
+        }
+        assertEquals(getMonitorAttrValue(baseDN, attributeName), expected, message);
+      }
+    });
+  }
+
+  /**
+   * Sets how many replay threads this server runs, the way an administrator would: the
+   * pool is stopped and created again with that number.
+   */
+  private static void setNumUpdateReplayThreads(int threads) throws Exception
+  {
+    assertEquals(TestCaseUtils.applyModifications(true,
+        "dn: " + SYNCHRO_PLUGIN_DN,
+        "changetype: modify",
+        "replace: ds-cfg-num-update-replay-threads",
+        "ds-cfg-num-update-replay-threads: " + threads), 0,
+        "the number of replay threads could not be changed");
+  }
+
+  /**
+   * Puts the number of replay threads back to what this server computes for itself, which
+   * is what it runs with when the configuration carries no number of its own.
+   *
+   * @return the result code of the change, so that a finally can call this without an
+   *         assertion which would replace the failure it is on the way out of
+   */
+  private static int resetNumUpdateReplayThreads() throws Exception
+  {
+    return TestCaseUtils.applyModifications(true,
+        "dn: " + SYNCHRO_PLUGIN_DN,
+        "changetype: modify",
+        "delete: ds-cfg-num-update-replay-threads");
+  }
+
+  /** Adds the entry a change of these tests is made on. */
+  private DN addEntryForChange(String uid) throws Exception
+  {
+    return TestCaseUtils.addEntry(
+        "dn: uid=" + uid + "," + baseDN,
+        "objectClass: top",
+        "objectClass: person",
+        "objectClass: organizationalPerson",
+        "objectClass: inetOrgPerson",
+        "uid: " + uid,
+        "cn: Aaccf Amar",
+        "sn: Amar").getName();
+  }
+
   /** A delivery of a change whose replay does not run to its end. */
   private interface FailingDelivery
   {
@@ -3603,16 +4106,7 @@
   private void assertChangeIsDeliveredAgainAfter(
       FailingDelivery delivery, int serverId, String uid, final String description) throws Exception
   {
-    Entry tmp = TestCaseUtils.addEntry(
-        "dn: uid=" + uid + "," + baseDN,
-        "objectClass: top",
-        "objectClass: person",
-        "objectClass: organizationalPerson",
-        "objectClass: inetOrgPerson",
-        "uid: " + uid,
-        "cn: Aaccf Amar",
-        "sn: Amar");
-    final DN dn = tmp.getName();
+    final DN dn = addEntryForChange(uid);
     final String uuid = getEntry(dn, 1, true).parseAttribute("entryuuid").asString();
 
     final LDAPReplicationDomain domain = MultimasterReplication.findDomain(baseDN, null);
@@ -3842,6 +4336,36 @@
   }
 
   /**
+   * A ModifyMsg whose ack runs out of memory on the way out of a replay which applied it.
+   * <p>
+   * The change is committed before the ack of its delivery is published, and the ack is
+   * where the JVM runs out of memory: the one throw from there which is not caught, so the
+   * replay is unwound with the change it was replaying in the data and owned by nobody -
+   * commit() cleared the owner - and the thread ends on the error. What the give-back on
+   * the way out of {@code replay()} has left to give back is the changes this thread parked
+   * as waiting for another one, and the restart it asks for them is the one which is run:
+   * the road a failed replay takes to ask for its own change again is not on the way.
+   * <p>
+   * Nothing on the way in reads what throws here: a message handed to the domain rather
+   * than published is not one this server acknowledges to anybody.
+   */
+  private static final class ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied extends ModifyMsg
+  {
+    private ModifyMsgWhoseAckRunsOutOfMemoryOnceApplied(
+        CSN csn, DN dn, List<Modification> mods, String entryUUID)
+    {
+      super(csn, dn, mods, entryUUID);
+    }
+
+    @Override
+    public boolean isAssured()
+    {
+      // Read first thing by processUpdateDone(), which is what publishes the ack.
+      throw new OutOfMemoryError("the ack of this applied delivery runs out of memory");
+    }
+  }
+
+  /**
    * An AddMsg whose ack throws on the way out of a replay which applied it.
    * <p>
    * The change is committed before the ack of its delivery is published, so this is the

--
Gitblit v1.10.0