[#1252] Go back to the changelog when the queue of a following handler drop...

Valery Kharseko

2 hours ago d27c892c81f6df051f737f54298cbdb530881d35

[#1252] Go back to the changelog when the queue of a following handler dropped changes while its writer waited on it (#1253)

Fixes #1252

## Problem

A replication server can skip a run of changes for good, whether it
sends them to a peer replication server or to a directory server.

`docker-convergence (rs534, none)` of run
[40832432665](https://github.com/OpenIdentityPlatform/OpenDJ/actions/runs/40832432665)
showed it with nothing disturbing the topology. The changelog of `cv-r0`
lacked 218 changes of `cv-d1` which `cv-r1` held. The directory servers
behind `cv-r0` kept four entries the others had deleted. #1252 has the
analysis of that run.

The writer of a handler which follows its in-memory queue waits on the
queue while it is empty, and gives up the monitor of the queue while it
waits (`MessageHandler.getNextMessage()`):
1. More changes than the queue holds can arrive before the writer has
the monitor back. `MessageHandler.add()` then drops the oldest of them
and sets `following` to false.
2. Back from the wait, the writer looked only at whether the queue was
empty. It took the head of the queue, the first change after the dropped
ones, and `updateServerState()` moved the state of the consumer past
them.
3. The read of the changelog the handler went back to then started after
the dropped changes, and `consumerHasWhatTheQueueLacks()` found them
covered by that state. They were never sent.

The `removeFirst()` right after the wait predates #963, which only added
the `continue` on an empty queue. `DataServerHandler` runs the same
code.

## Fix

Back from the wait, the writer goes back to the changelog when the
handler stopped following, as it already did when the queue was empty.
`fillLateQueue()` reads from the state of the consumer, which does not
hold the dropped changes, and `queueLacksUpTo` keeps the handler off the
queue until the consumer has them (#1206).

## Tests


`MessageHandlerCatchUpTest.changesTheQueueDroppedWhileTheWriterWaitedOnItAreSent`:
- The handler has a queue of 10 and an empty changelog. Its writer finds
nothing to catch up on, follows the queue and waits on it.
- The test holds the monitor of the queue while 15 changes arrive
(written to the changelog, then handed to the queue), then lets the
writer go.
- The consumer must be sent all 15, in order.

Nothing else holds the writer off the monitor on cue, so the test takes
the monitor itself, through the private field of the queue. It checks
that the queue did stop following, so the case cannot pass without the
queue having dropped changes.

## Verification

- **Without the fix:** the new case fails. The consumer is sent seq 6 to
15, and the 5 changes the queue dropped are never sent, as in the CI
run. The 12 other cases of `MessageHandlerCatchUpTest` and the 8 of
`MissedUpdateRecoveryTest` pass.
- **With the fix:** 13 of 13 and 8 of 8 pass.

 opendj-server-legacy/src/main/java/org/opends/server/replication/server/MessageHandler.java

@@ -513,9 +513,16 @@
          {
            return null;
          }
          if (msgQueue.isEmpty())
          if (msgQueue.isEmpty() || !following)
          {
            // this handler is going back to the changelog, see stopFollowingWhenChangesAreMissing()
            /*
             * This handler is going back to the changelog: changes are missing, see
             * stopFollowingWhenChangesAreMissing(), or the queue dropped its oldest changes in
             * add() while this thread waited without the monitor - more changes arrived than
             * it holds. Its head is then the first change after the dropped ones: taken now, it
             * would move the state of the consumer past them, and the changelog would be read
             * from after them, so they would never be sent - see issue #1252.
             */
            continue;
          }
          UpdateMsg msg = msgQueue.removeFirst();

 opendj-server-legacy/src/test/java/org/opends/server/replication/server/MessageHandlerCatchUpTest.java

@@ -20,6 +20,7 @@
import static org.opends.server.TestCaseUtils.TEST_ROOT_DN_STRING;
import static org.opends.server.util.CollectionUtils.newArrayList;

import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;
import java.util.TreeSet;
@@ -516,6 +517,82 @@
  }

  /**
   * The case of #1252: the writer of a handler which follows its queue waits on it, empty, and
   * more changes arrive than the queue holds before the writer has the monitor back. The queue
   * drops its oldest changes and stops following; the writer must then go back to the changelog
   * for them rather than take the head of the queue, which is the first change after them.
   * <p>
   * Nothing holds the writer off the monitor on cue but the monitor itself, so the test takes it
   * while the changes arrive, as the threads which add changes kept it from the writer in the run
   * of #1252.
   */
  @Test
  public void changesTheQueueDroppedWhileTheWriterWaitedOnItAreSent() throws Exception
  {
    final DN baseDN = DN.valueOf(TEST_ROOT_DN_STRING);
    ReplicationServer replicationServer = null;
    final ExecutorService writer = Executors.newSingleThreadExecutor();
    try
    {
      replicationServer = newReplicationServer("messageHandlerWaitingDb");
      final List<UpdateMsg> changesOfA = changesOfA(baseDN, TimeThread.getTime() - 60000, 1, 15);
      final MessageHandler handler = newHandler(replicationServer, baseDN, 10, new ServerState());

      // the changelog is empty: the writer has nothing to catch up on, follows the queue and waits on it
      final Future<Thread> writerThread = writer.submit(Thread::currentThread);
      final Future<UpdateMsg> first = writer.submit(handler::getNextMessage);
      waitUntilItWaitsOnItsQueue(handler, writerThread.get());

      synchronized (queueOf(handler))
      {
        for (UpdateMsg change : changesOfA)
        {
          arrive(replicationServer, handler, change);
        }
        assertThat(handler.isFollowing())
            .as("the queue did not drop any change: the test shows nothing")
            .isFalse();
      }

      final List<UpdateMsg> sent = new ArrayList<>();
      sent.add(first.get(NEXT_MESSAGE_TIMEOUT_MS, TimeUnit.MILLISECONDS));
      for (Sent next : nextMessages(handler, changesOfA.size() - 1))
      {
        sent.add(next.msg);
      }
      assertThat(csnsOf(sent))
          .as("the consumer was not sent every change, in order")
          .containsExactlyElementsOf(csnsOf(changesOfA));
    }
    finally
    {
      writer.shutdownNow();
      removeQuietly(replicationServer);
    }
  }

  /** Waits until the writer waits on the queue of the handler, which it then follows. */
  private static void waitUntilItWaitsOnItsQueue(MessageHandler handler, Thread writer) throws Exception
  {
    final long deadline = System.currentTimeMillis() + NEXT_MESSAGE_TIMEOUT_MS;
    while (!handler.isFollowing() || writer.getState() != Thread.State.TIMED_WAITING)
    {
      assertThat(System.currentTimeMillis())
          .as("the writer did not come to wait on the queue it follows")
          .isLessThan(deadline);
      Thread.sleep(10);
    }
  }

  /** The queue of the handler, the monitor of which its writer waits on. */
  private static Object queueOf(MessageHandler handler) throws Exception
  {
    final Field queue = MessageHandler.class.getDeclaredField("msgQueue");
    queue.setAccessible(true);
    return queue.get(handler);
  }

  /**
   * A queue of {@link #size} changes which drops a change the consumer lacks, then takes a change
   * of B with an older CSN in the room the drop left - which only a queue bounded by its size in
   * bytes leaves, as it drops whole changes: one large change, dropped, leaves room for a small one.