[#1227] Keep the record a changelog cursor read in advance when the head lo...

Valery Kharseko

23 hours ago 53c6d33b0c2e612aff1352a8f237ee6e83cdfe42

[#1227] Keep the record a changelog cursor read in advance when the head log file it is positioned in is rotated (#1228)

Fixes #1227

## Problem

A replication server can skip one change of a replica for a directory
server which is catching up, and the state of that directory server then
moves past the change for good. This is the one-change loss of #1225,
which `docker-convergence (combined)` showed again in run
[39407529872](https://github.com/OpenIdentityPlatform/OpenDJ/actions/runs/39407529872)
on a build with #1219. #1227 has the analysis of that run.

A cursor on a changelog log reads the record it is positioned on in
advance:
1. `LogFileCursor.positionTo()` keeps that record in `initialRecord` for
the next `next()`, and leaves the file pointer after it.
2. When the head log file is rotated, `Log.rotateHeadLogFile()` moves
every cursor open on it to the read-only file it was renamed to. It does
so from the state `InternalLogCursor.getState()` saved: the file
position and the current record, but not `initialRecord`.
3. A cursor positioned on the head log file and moved before it read
anything therefore starts after the record it is positioned on.
`FileReplicaDBCursor` takes the next one for the first change after its
start CSN.

`positionTo()` and the first `next()` take the shared lock of the log
one after the other. So a rotation which asks for the exclusive lock in
between runs before the read. Two roads go through that window:
- `MessageHandler.fillLateQueue()` makes such cursors anew for every
batch it reads.
-
`FileReplicaDBCursor.nextWhenCursorIsExhaustedOrNotCorrectlyPositionned()`
positions again each time it looks for the changes written since.

## Fix

- `Log.CursorState` keeps the record read in advance, and the state
moved to the rotated file carries it.
- `LogFile.LogFileCursor.getInitialRecord()` gives that record to the
state.
- `LogFile.getCursorInitialisedTo()` gives it back to the cursor made on
the rotated file.

## Tests

In `LogTest`, on its log which rotates after two records:
-
`aCursorReadsTheRecordItIsPositionedOnWhenTheHeadLogFileIsRotatedBeforeItsFirstRead`:
a cursor positioned on the head log file reads that record first after a
rotation. It covers every key matching and positioning strategy which
reads a record in advance, 6 cases.
-
`aCursorPositionedAgainReadsTheRecordItIsPositionedOnWhenTheHeadLogFileIsRotatedBeforeItsNextRead`:
the same for a cursor which read the newest record and is positioned
again once more were written. This is the road `FileReplicaDBCursor`
takes.

## Verification

- **Without the fix:** 7 of the 95 tests of `LogTest` fail, each one on
the skipped record (expected `key009`, `key010` or `key012`, read the
record after it).
- **With the fix:** 95 of 95 pass.
- **Mutants:** each link of the fix was broken in turn, 5 mutants:
  - the state taken before the rotation drops the record;
  - the state moved to the rotated file drops it;
  - the cursor made from the state is not given it;
  - the cursor does not keep it;
  - the getter returns the current record instead.

  Each one fails all 7 new cases.
- **Reactor** (`-Pprecommit verify -pl opendj-server-legacy -am`, JDK
27, with javadoc), `org/opends/server/replication/**` and
`ChangelogBackendTestCase`:
- 3967 tests, 2 failures. Both failures are the embedded server failing
to start in `setUp` of `FileReplicaDBTest` and `ComputeBestServerTest`,
on `0.0.0.0:65534: Address already in use`: another test server held the
admin port.
  - Run again on their own, both classes pass (17 and 46 tests).
- Every class of `replication/server/changelog/file` passes, and
`ChangelogBackendTestCase` passes (31).

 opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/Log.java

@@ -1048,7 +1048,8 @@
        final K previousKey = logFiles.lowerKey(recordParser.getMaxKey());
        final LogFile<K, V> logFile = findLogFileFor(previousKey, KeyMatchingStrategy.EQUAL_TO_KEY);
        final AbortableLogCursor<K, V> cursor = pair.getFirst();
        cursor.reinitializeTo(new CursorState<K, V>(logFile, cursorState.filePosition, cursorState.record));
        cursor.reinitializeTo(new CursorState<K, V>(
            logFile, cursorState.filePosition, cursorState.record, cursorState.initialRecord));
      }
    }
  }
@@ -1301,7 +1302,8 @@
      log.sharedLock.lock();
      try
      {
        return new CursorState<>(currentLogFile, currentCursor.getFilePosition(), currentCursor.getRecord());
        return new CursorState<>(currentLogFile, currentCursor.getFilePosition(), currentCursor.getRecord(),
            currentCursor.getInitialRecord());
      }
      finally
      {
@@ -1320,7 +1322,8 @@
    void reinitializeTo(final CursorState<K, V> cursorState) throws ChangelogException
    {
      currentLogFile = cursorState.logFile;
      currentCursor = currentLogFile.getCursorInitialisedTo(cursorState.record, cursorState.filePosition);
      currentCursor = currentLogFile.getCursorInitialisedTo(
          cursorState.record, cursorState.initialRecord, cursorState.filePosition);
    }

    @Override
@@ -1605,6 +1608,14 @@
    /** The record the cursor is pointing to. */
    private final Record<K,V> record;

    /**
     * The record the cursor read in advance when it was positioned, and returns before it reads at
     * {@link #filePosition}, which is past it. Without it, a cursor positioned on the head log file
     * and moved to the rotated file before it read anything would skip the record it is positioned
     * on - see issue #1227.
     */
    private final Record<K, V> initialRecord;

    private final boolean isValid;

    /** Creates a non-valid state. */
@@ -1612,15 +1623,18 @@
      logFile = null;
      filePosition = 0;
      record = null;
      initialRecord = null;
      isValid = false;
    }

    /** Creates a valid state. */
    private CursorState(final LogFile<K, V> logFile, final long filePosition, final Record<K, V> record)
    private CursorState(final LogFile<K, V> logFile, final long filePosition, final Record<K, V> record,
        final Record<K, V> initialRecord)
    {
      this.logFile = logFile;
      this.filePosition = filePosition;
      this.record = record;
      this.initialRecord = initialRecord;
      isValid = true;
    }


 opendj-server-legacy/src/main/java/org/opends/server/replication/server/changelog/file/LogFile.java

@@ -368,15 +368,19 @@
   *
   * @param record
   *            The initial record this cursor points on
   * @param initialRecord
   *            The record the cursor read in advance, which it returns before it reads at the
   *            provided position, or {@code null} if there is none
   * @param position
   *            The file position this cursor points on
   * @return the cursor
   * @throws ChangelogException
   *            If a problem occurs while creating the cursor.
   */
  LogFileCursor<K, V> getCursorInitialisedTo(Record<K,V> record, long position) throws ChangelogException
  LogFileCursor<K, V> getCursorInitialisedTo(Record<K,V> record, Record<K, V> initialRecord, long position)
      throws ChangelogException
  {
    return new LogFileCursor<>(this, record, position);
    return new LogFileCursor<>(this, record, initialRecord, position);
  }

  /**
@@ -583,17 +587,19 @@
    }

    /**
     * Creates a cursor on the provided log, initialised to the provided record and
     * Creates a cursor on the provided log, initialised to the provided records and
     * pointing to the provided file position.
     * <p>
     * Note: there is no check to ensure that provided record and file position are
     * Note: there is no check to ensure that provided records and file position are
     * consistent. It is the responsibility of the caller of this method.
     */
    private LogFileCursor(LogFile<K, V> logFile, Record<K, V> record, long filePosition) throws ChangelogException
    private LogFileCursor(LogFile<K, V> logFile, Record<K, V> record, Record<K, V> initialRecord, long filePosition)
        throws ChangelogException
    {
      this.logFile = logFile;
      this.reader = logFile.getReader();
      this.currentRecord = record;
      this.initialRecord = initialRecord;
      logFile.sharedLock.lock();
      try
      {
@@ -635,6 +641,17 @@
      return currentRecord;
    }

    /**
     * Returns the record this cursor read in advance when it was positioned, which the next call
     * to {@link #next()} returns before anything is read from the file.
     *
     * @return the record read in advance, or {@code null} if there is none
     */
    Record<K, V> getInitialRecord()
    {
      return initialRecord;
    }

    /** {@inheritDoc} */
    @Override
    public boolean positionTo(final K key, final KeyMatchingStrategy match, final PositionStrategy pos)

 opendj-server-legacy/src/test/java/org/opends/server/replication/server/changelog/file/LogTest.java

@@ -32,6 +32,7 @@
import org.opends.server.replication.server.changelog.api.DBCursor.KeyMatchingStrategy;
import org.opends.server.replication.server.changelog.api.DBCursor.PositionStrategy;
import org.opends.server.replication.server.changelog.file.Log.LogRotationParameters;
import org.opends.server.replication.server.changelog.file.Log.RepositionableCursor;
import org.opends.server.replication.server.changelog.file.LogFileTest.FailingStringRecordParser;
import org.opends.server.replication.server.changelog.file.Record.Mapper;
import org.opends.server.util.StaticUtils;
@@ -360,6 +361,69 @@
    }
  }

  @DataProvider
  Object[][] positionedOnTheHeadLogFile()
  {
    return new Object[][] {
      // key to position to, key matching strategy, position strategy, index of the first record read
      { "key009", EQUAL_TO_KEY, ON_MATCHING_KEY, 9 },
      { "key009", EQUAL_TO_KEY, AFTER_MATCHING_KEY, 10 },
      { "key009", GREATER_THAN_OR_EQUAL_TO_KEY, ON_MATCHING_KEY, 9 },
      { "key009", GREATER_THAN_OR_EQUAL_TO_KEY, AFTER_MATCHING_KEY, 10 },
      // key is between key009 and key010: the record before it is read in advance
      { "key0095", LESS_THAN_OR_EQUAL_TO_KEY, ON_MATCHING_KEY, 9 },
      // key is after the newest key: the newest record is read in advance
      { "key0105", LESS_THAN_OR_EQUAL_TO_KEY, ON_MATCHING_KEY, 10 },
    };
  }

  /**
   * A cursor positioned on a key has read the record it is positioned on in advance, and is past it
   * in the file: when the head log file it is positioned in is rotated before it reads anything, it
   * must still read that record first (issue #1227).
   */
  @Test(dataProvider = "positionedOnTheHeadLogFile")
  public void aCursorReadsTheRecordItIsPositionedOnWhenTheHeadLogFileIsRotatedBeforeItsFirstRead(String key,
      KeyMatchingStrategy matchingStrategy, PositionStrategy positionStrategy, int firstIndex) throws Exception
  {
    try (Log<String, String> log = openLog(LogFileTest.RECORD_PARSER);
        DBCursor<Record<String, String>> cursor = log.getCursor(key, matchingStrategy, positionStrategy))
    {
      // the head log file holds 2 records: this append rotates it
      log.append(Record.from("key011", "value11"));

      assertThatCursorCanBeFullyReadFromStart(cursor, firstIndex, 11);
    }
  }

  /**
   * A cursor which read the newest record is positioned again to read the records written since - the
   * way FileReplicaDBCursor does - and reads the record it is positioned on in advance as well: when
   * the head log file is rotated before its next read, it must still read that record first (issue
   * #1227).
   */
  @Test
  public void aCursorPositionedAgainReadsTheRecordItIsPositionedOnWhenTheHeadLogFileIsRotatedBeforeItsNextRead()
      throws Exception
  {
    try (Log<String, String> log = openLog(LogFileTest.RECORD_PARSER))
    {
      // the head log file holds 2 records: this append rotates it, and the new one holds key011
      log.append(Record.from("key011", "value11"));
      try (RepositionableCursor<String, String> cursor = log.getCursor())
      {
        assertThatCursorCanBeFullyReadFromStart(cursor, 1, 11);
        log.append(Record.from("key012", "value12"));

        assertThat(cursor.positionTo("key011", GREATER_THAN_OR_EQUAL_TO_KEY, AFTER_MATCHING_KEY)).isTrue();
        // the head log file holds 2 records: this append rotates it
        log.append(Record.from("key013", "value13"));

        assertThatCursorCanBeFullyRead(cursor, 12, 13);
      }
    }
  }

  @Test
  public void testClear() throws Exception
  {