[#1256] Remove incomplete changelog records completely during crash recover...

Valery Kharseko

16 hours ago c3e63df192c7ea16efeeda75a0603de2b617cf65

[#1256] Remove incomplete changelog records completely during crash recovery (#1257)

Fixes #1256

After a replication server is killed during a changelog write, recovery
could truncate at the last block boundary while retaining the beginning
of an incomplete record. Subsequent appends then corrupted record
framing. In the failing `rs534, restart-rs` run of #1244, the server
reported successful recovery but later failed to decode the changelog
and repeatedly closed a directory server's connection.

`BlockLogReader.checkLogIsValid()` now uses the record start reached
through the block offset as the initial safe truncation position. For a
zero offset, it retains the block boundary so that the offset field is
removed with the incomplete record. If the crash interrupted the offset
itself, recovery starts from the preceding complete block. Errors
decoding a complete record remain fatal.

## Regression coverage

Thirteen `LogFileTest` cases exercise recovery through the normal
`LogFile` open path. They cover incomplete bodies across one or several
blocks, length headers split across blocks, zero offsets, exact block
boundaries, partially written offsets, and an incomplete first record.
Each checks the recovered length and newest record, appends a new
record, then reopens the file and reads the complete expected sequence.

On the original implementation, ten cases fail: six on the wrong
truncation position and four on EOF while reading an incomplete offset.
The other three cases and all 52 existing `LogFileTest` cases pass. With
the fix, all 65 pass.

## Validation

JDK 21, offline Maven reactor:

```sh
mvn -o -Pprecommit verify -pl opendj-server-legacy -am \
  '-Dit.test=**/replication/server/changelog/file/*Test,MessageHandlerCatchUpTest,MissedUpdateRecoveryTest' \
  -Dfailsafe.failIfNoSpecifiedTests=false \
  -Dtest=NoTestsSelected -Dsurefire.failIfNoSpecifiedTests=false
```

- Reactor verify succeeds, including Javadoc: **1,457 integration tests
across 13 classes**, zero failures, errors or skips. The command selects
the file-changelog and handler-recovery tests and skips unrelated
Surefire tests.
- Before the change, the baseline reactor verify passed all 1,346
selected integration tests in `BlockLogReaderWriterTest`, `LogFileTest`
and `LogTest`, with upstream Surefire tests enabled.
- The standalone reproduction from #1256 now returns the correct
truncation position, **104**, instead of **256**.
- `git diff --check` passes.

The Docker `rs534, restart-rs` scenario was not rerun locally. This
change addresses the file-recovery defect in #1256; it does not claim to
resolve every other symptom in the CI run discussed there.

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

@@ -12,6 +12,7 @@
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2014-2016 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.server.changelog.file;

@@ -600,10 +601,17 @@
 {
   try
   {
     final long lastBlockStart = getClosestBlockStartToEndOfFile();
     long lastBlockStart = getClosestBlockStartToEndOfFile();
     if (lastBlockStart > 0 && getFileLength() - lastBlockStart < SIZE_OF_BLOCK_OFFSET)
     {
       // A crash may interrupt the block offset itself. Read from the preceding complete block.
       lastBlockStart -= blockSize;
     }
     positionToRecordFromBlockStart(lastBlockStart);

     long lastValidPosition = lastBlockStart;
     // The first record may be incomplete and start before this block: remove it in its entirety.
     // With a zero offset the reader skipped the offset field, which must be removed with the record.
     long lastValidPosition = Math.min(lastBlockStart, reader.getFilePointer());
     for (ByteString recordData = readNextRecord(); recordData != null; recordData = readNextRecord()) {
       parser.decodeRecord(recordData);
       lastValidPosition = reader.getFilePointer();

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

@@ -12,6 +12,7 @@
 * information: "Portions Copyright [year] [name of copyright owner]".
 *
 * Copyright 2014-2016 ForgeRock AS.
 * Portions Copyright 2026 3A Systems, LLC.
 */
package org.opends.server.replication.server.changelog.file;

@@ -252,6 +253,79 @@
    }
  }

  @DataProvider
  Object[][] interruptedRecordData()
  {
    return new Object[][]
    {
      // First value length, complete prefix length, interrupted file length (256-byte blocks).
      { 93, 104, 350 },  // incomplete body crossing a block
      { 93, 104, 700 },  // incomplete body crossing several blocks
      { 242, 253, 300 }, // length header split across a block, as in issue #1256
      { 494, 509, 600 }, // split length header after a complete record spanning blocks
      { 245, 256, 260 }, // zero block offset, with no record bytes after it
      { 245, 256, 270 }, // incomplete record starting at a block boundary
      { 93, 104, 256 },  // incomplete body ending exactly at a block boundary
      { 93, 104, 260 },  // complete block offset, but no continuation bytes
      { 93, 104, 257 },  // the block offset itself is incomplete
      { 93, 104, 258 },
      { 93, 104, 259 },
      { 245, 256, 257 }, // incomplete offset following a complete record
      { 0, 0, 300 },     // the first record is incomplete
    };
  }

  @Test(dataProvider="interruptedRecordData")
  public void recoveryRemovesTheWholeInterruptedRecordBeforeAppending(
      int firstValueLength, long completePrefixLength, long interruptedLength) throws Exception
  {
    try (RandomAccessFile output = new RandomAccessFile(TEST_LOG_FILE, "rw"))
    {
      output.setLength(0);
    }
    final Record<String, String> first =
        Record.from("key01", new String(new char[firstValueLength]).replace('\0', 'a'));
    final Record<String, String> interrupted =
        Record.from("key02", new String(new char[900]).replace('\0', 'b'));
    final Record<String, String> appended = Record.from("key03", "after recovery");
    try (LogFile<String, String> logFile = getLogFile(RECORD_PARSER))
    {
      if (completePrefixLength > 0)
      {
        assertThat(logFile.append(first)).isTrue();
      }
      assertThat(TEST_LOG_FILE.length()).isEqualTo(completePrefixLength);
      assertThat(logFile.append(interrupted)).isTrue();
    }
    assertThat(TEST_LOG_FILE.length()).isGreaterThan(interruptedLength);
    try (RandomAccessFile output = new RandomAccessFile(TEST_LOG_FILE, "rw"))
    {
      output.setLength(interruptedLength);
    }

    try (LogFile<String, String> logFile = getLogFile(RECORD_PARSER))
    {
      assertThat(TEST_LOG_FILE.length()).as("the entire incomplete record must be removed")
          .isEqualTo(completePrefixLength);
      assertThat(logFile.getNewestRecord()).isEqualTo(completePrefixLength > 0 ? first : null);
      assertThat(logFile.append(appended)).isTrue();
    }
    // A second open must preserve both the complete prefix and the newly appended record.
    try (LogFile<String, String> logFile = getLogFile(RECORD_PARSER);
         DBCursor<Record<String, String>> cursor = logFile.getCursor())
    {
      assertThat(logFile.getNewestRecord()).isEqualTo(appended);
      if (completePrefixLength > 0)
      {
        assertThat(cursor.next()).isTrue();
        assertThat(cursor.getRecord()).isEqualTo(first);
      }
      assertThat(cursor.next()).isTrue();
      assertThat(cursor.getRecord()).isEqualTo(appended);
      assertThatCursorIsExhausted(cursor);
    }
  }

  /** Append some raw data to the TEST_LOG_FILE. Intended to corrupt the log file. */
  private void corruptTestLogFile(ByteStringBuilder corruptedRecordData) throws Exception
  {