[#861] Fail fast when a total update request gets no answer (#864)

Valery Kharseko

18 hours ago 641ff2ce8441a33717bdf066c3b0ae2fe96ba271

[#861] Fail fast when a total update request gets no answer (#864)

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

@@ -13,8 +13,7 @@
 *
 * Copyright 2008-2010 Sun Microsystems, Inc.
 * Portions Copyright 2011-2016 ForgeRock AS.
 * Portions Copyright 2025-2026 3A Systems LLC.
 * Portions Copyright 2026 3A Systems, LLC.
 * Portions Copyright 2025-2026 3A Systems, LLC.
 */
package org.opends.server.replication.service;

@@ -821,9 +820,12 @@
                  " Error Msg received: " + errorMsg);
            }

            if (errorMsg.getCreationTime() > ieCtx.startTime)
            // consider only ErrorMsg that relate to the current import/export.
            // ">=" and not ">": with all servers on one host the whole
            // request/rejection round-trip can complete within the millisecond
            // this context was created in (issue #861)
            if (errorMsg.getCreationTime() >= ieCtx.startTime)
            {
              // consider only ErrorMsg that relate to the current import/export
              processErrorMsg(errorMsg, ieCtx);
            }
            else
@@ -1146,11 +1148,28 @@
    private InitializeRequestMsg initReqMsgSent;

    /**
     * Start time of the initialization process. ErrorMsg timestamped before
     * this startTime will be ignored.
     * Start time of the initialization process. ErrorMsg timestamped strictly
     * before this startTime will be ignored.
     */
    private final long startTime;

    /**
     * Time when {@link #initReqMsgSent} was last published. Volatile: written
     * by the requesting and listener threads, read by the task thread running
     * {@link ReplicationDomain#abortStalledInitializeFromRemote(long)}.
     */
    private volatile long requestSentTime;
    /**
     * Whether the InitializeTargetMsg answering {@link #initReqMsgSent} has
     * been received. Guarded by this context's monitor.
     */
    private boolean startReceived;
    /**
     * Whether the stalled-request watchdog abandoned this context. Guarded by
     * this context's monitor.
     */
    private boolean abandonedAsStalled;

    /** List for replicas (DS) connected to the topology when initialization started. */
    private final Set<Integer> startList = new HashSet<>(0);

@@ -1202,6 +1221,58 @@
    }

    /**
     * Returns the start time of this initialization, for tests.
     *
     * @return the creation time of this context in milliseconds
     */
    long getStartTime()
    {
      return startTime;
    }

    /** Arms (or re-arms, on a new attempt) the stalled-request watchdog. */
    private synchronized void markInitRequestSent()
    {
      startReceived = false;
      requestSentTime = System.currentTimeMillis();
    }

    /**
     * Marks that the InitializeTargetMsg answering the published
     * InitializeRequestMsg has been received.
     *
     * @return false when the stalled-request watchdog already abandoned this
     *         context, in which case the start message must be ignored
     */
    private synchronized boolean markInitStartReceived()
    {
      if (abandonedAsStalled)
      {
        return false;
      }
      startReceived = true;
      return true;
    }

    /**
     * Abandons this context when the published request has received no answer
     * within the provided delay.
     *
     * @param stalledTimeoutMs delay after which the request is considered lost
     * @return whether this call abandoned the context
     */
    private synchronized boolean abandonIfStalled(long stalledTimeoutMs)
    {
      if (startReceived || abandonedAsStalled
          || System.currentTimeMillis() - requestSentTime < stalledTimeoutMs)
      {
        return false;
      }
      abandonedAsStalled = true;
      return true;
    }

    /**
     * Returns the total number of entries to be processed when a total update
     * is in progress.
     *
@@ -1326,7 +1397,10 @@
     */
    public void setExceptionIfNoneSet(DirectoryException exception)
    {
      if (exception == null)
      // Historical upstream bug (since at least OpenDJ 3): the null check was
      // made on the argument instead of the field, so no error was ever
      // recorded and every failed total update completed "successfully"
      if (this.exception == null)
      {
        this.exception = exception;
      }
@@ -2064,7 +2138,7 @@
          if (ieCtx.getException() == null)
          {
            ErrorMsg errMsg = (ErrorMsg)msg;
            if (errMsg.getCreationTime() > ieCtx.startTime)
            if (errMsg.getCreationTime() >= ieCtx.startTime)
            {
              ieCtx.setException(
                  new DirectoryException(ResultCode.OTHER,errMsg.getDetails()));
@@ -2308,7 +2382,16 @@
      ieCtx.attemptCnt = 0;
      ieCtx.initReqMsgSent = new InitializeRequestMsg(
          getBaseDN(), getServerId(), source, getInitWindow());
      broker.publish(ieCtx.initReqMsgSent);
      ieCtx.markInitRequestSent();
      // The broker silently drops the message when it is caught between two
      // sessions (connection error, recovery pending after a reconnect) and
      // only replays UpdateMsgs on reconnect: an unpublished request would
      // leave the task waiting forever for an answer (issue #861).
      if (!broker.publish(ieCtx.initReqMsgSent, true))
      {
        throw new DirectoryException(ResultCode.OTHER,
            ERR_INITIALIZATION_FAILED_NOCONN.get(getBaseDN()));
      }

      /*
      The normal success processing is now to receive InitTargetMsg then
@@ -2340,6 +2423,42 @@
  }

  /**
   * Fails the on-going initialization from a remote replica when the request
   * published by {@link #initializeFromRemote(int, Task)} has received no
   * answer at all - neither the InitializeTargetMsg starting the import nor an
   * ErrorMsg - within the provided delay. The request or its answer can be
   * lost with no error ever coming back (issue #861), and nothing else bounds
   * the wait: without this watchdog the initialize task hangs forever.
   *
   * @param stalledTimeoutMs
   *          delay in milliseconds after which the unanswered request is
   *          considered lost
   * @return whether a stalled initialization was aborted by this call
   */
  public boolean abortStalledInitializeFromRemote(long stalledTimeoutMs)
  {
    final ImportExportContext ieCtx = importExportContext.get();
    if (ieCtx == null || !ieCtx.importInProgress() || ieCtx.initReqMsgSent == null
        || !ieCtx.abandonIfStalled(stalledTimeoutMs))
    {
      return false;
    }
    // Once abandonIfStalled() returned true a concurrently received
    // InitializeTargetMsg is ignored by the listener, so releasing the
    // context here cannot race the start of an import.
    ieCtx.setExceptionIfNoneSet(new DirectoryException(ResultCode.OTHER,
        ERR_NO_REACHABLE_PEER_IN_THE_DOMAIN.get(
            getBaseDN(), ieCtx.initReqMsgSent.getDestination())));
    if (ieCtx.initializeTask instanceof InitializeTask)
    {
      ((InitializeTask) ieCtx.initializeTask)
          .updateTaskCompletionState(ieCtx.getException());
    }
    releaseIEContext();
    return true;
  }

  /**
   * Processes an InitializeTargetMsg received from a remote server
   * meaning processes an initialization from the entries expected to be
   * received now.
@@ -2359,8 +2478,48 @@
    }

    InitializeTask initFromTask = null;
    int source = initTargetMsgReceived.getSenderID();
    ImportExportContext ieCtx = importExportContext.get();
    final int source = initTargetMsgReceived.getSenderID();
    final ImportExportContext ieCtx;
    if (initTargetMsgReceived.getInitiatorID() == getServerId())
    {
      ieCtx = importExportContext.get();
      if (ieCtx == null || !ieCtx.markInitStartReceived())
      {
        /*
        The stalled-request watchdog abandoned the initialization this message
        answers (issue #861): its task already failed and its context is (about
        to be) released. The entries following this message are discarded by
        the listener until the exporter completes.
        */
        if (logger.isTraceEnabled())
        {
          logger.trace("[IE] Ignoring InitializeTargetMsg from server " + source
              + " for domain " + getBaseDN()
              + ": the initialization was abandoned as stalled");
        }
        return;
      }
    }
    else
    {
      /*
      The initTargetMsgReceived is for an import initiated by the remote
      server. Test and set if no import already in progress
      */
      try
      {
        ieCtx = acquireIEContext(true);
      }
      catch (DirectoryException e)
      {
        // A concurrent import/export owns the context: reject this
        // initialization without touching that operation's context, and let
        // the exporter know so that it does not export to a replica that
        // will discard the entries
        broker.publish(new ErrorMsg(requesterServerId, e.getMessageObject()));
        return;
      }
    }
    try
    {
      // Log starting
@@ -2370,16 +2529,6 @@
      // Go into full update status
      setNewStatus(StatusMachineEvent.TO_FULL_UPDATE_STATUS_EVENT);

      // Acquire an import context if no already done (and initialize).
      if (initTargetMsgReceived.getInitiatorID() != getServerId())
      {
        /*
        The initTargetMsgReceived is for an import initiated by the remote server.
        Test and set if no import already in progress
        */
        ieCtx = acquireIEContext(true);
      }

      // Initialize stuff
      ieCtx.importSource = source;
      ieCtx.initializeCounters(initTargetMsgReceived.getEntryCount());
@@ -2435,7 +2584,15 @@
            logger.info(NOTE_RESENDING_INIT_FROM_REMOTE_REQUEST,
                ieCtx.getException().getLocalizedMessage());

            broker.publish(ieCtx.initReqMsgSent);
            ieCtx.markInitRequestSent();
            if (!broker.publish(ieCtx.initReqMsgSent, true))
            {
              // Same silent-drop hazard as the first request (issue #861):
              // fail the attempt instead of waiting for an answer that
              // cannot arrive
              throw new DirectoryException(ResultCode.OTHER,
                  ERR_INITIALIZATION_FAILED_NOCONN.get(getBaseDN()));
            }

            ieCtx.initializeCounters(0);
            ieCtx.exception = null;

 opendj-server-legacy/src/main/java/org/opends/server/tasks/InitializeTask.java

@@ -40,6 +40,14 @@
{
  private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();

  /**
   * Delay after which the initialization is considered stalled when the
   * request for entries has received no answer from the topology: the request
   * or its response can be silently lost with no error reported back (issue
   * #861), and no other mechanism bounds this wait.
   */
  private static final long INITIALIZE_START_TIMEOUT_MS = 2 * 60 * 1000L;

  private String domainString;
  private int  source;
  private LDAPReplicationDomain domain;
@@ -120,6 +128,7 @@
          initStateLock.wait(1000);
          replaceAttributeValue(ATTR_TASK_INITIALIZE_LEFT, String.valueOf(left));
          replaceAttributeValue(ATTR_TASK_INITIALIZE_DONE, String.valueOf(total-left));
          domain.abortStalledInitializeFromRemote(INITIALIZE_START_TIMEOUT_MS);
        }
      }
      replaceAttributeValue(ATTR_TASK_INITIALIZE_LEFT, String.valueOf(left));

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

@@ -27,6 +27,7 @@
import java.util.Map;
import java.util.SortedSet;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
@@ -47,6 +48,7 @@
import org.opends.server.replication.server.ReplicationServer;
import org.opends.server.replication.service.ReplicationDomain.ImportExportContext;
import org.forgerock.opendj.ldap.DN;
import org.opends.server.tasks.InitializeTask;
import org.opends.server.types.DirectoryException;
import org.opends.server.util.TestTimer;
import org.testng.annotations.DataProvider;
@@ -608,6 +610,154 @@
    }
  }

  /** InitializeTask double recording the completion reported by the domain. */
  private static final class RecordingInitializeTask extends InitializeTask
  {
    private final CountDownLatch completed = new CountDownLatch(1);
    private volatile DirectoryException failure;

    @Override
    public void updateTaskCompletionState(DirectoryException de)
    {
      failure = de;
      completed.countDown();
    }

    DirectoryException waitForCompletion(long timeout, TimeUnit unit) throws InterruptedException
    {
      assertTrue(completed.await(timeout, unit), "the initialize task never completed");
      return failure;
    }
  }

  /**
   * An ErrorMsg answering an initialization request can be created within the
   * same millisecond as the requester's import/export context when the whole
   * topology runs on one host: it must terminate the pending initialization
   * instead of being discarded as stale (issue #861).
   */
  @Test(enabled=true)
  public void errorMsgFromSameMillisecondTerminatesPendingInitialize() throws Exception
  {
    DN testService = DN.valueOf("o=test");
    ReplicationServer replServer = null;
    FakeReplicationDomain domain2 = null;
    ReplicationBroker broker3 = null;

    try
    {
      int replServerPort = TestCaseUtils.findFreePort();
      replServer = createReplicationServer(13, replServerPort,
          "sameMillisecondErrorMsgDb", 100);
      SortedSet<String> servers = newTreeSet("localhost:" + replServerPort);

      domain2 = new FakeReplicationDomain(
          testService, 2, servers, 0, null, new StringBuffer(), 0);
      broker3 = openReplicationSession(testService, 3, 100, replServerPort,
          10000, domain2.getGenerationID());

      waitTopologyKnowsReplica(domain2, 3);

      RecordingInitializeTask task = new RecordingInitializeTask();
      domain2.initializeFromRemote(3, task);
      long startTime = domain2.getImportExportContext().getStartTime();

      // strictly older than the context: still ignored as stale
      ErrorMsg staleError = new ErrorMsg(3, 2, LocalizableMessage.raw("stale error"));
      staleError.setCreationTime(startTime - 1);
      broker3.publish(staleError);

      // same millisecond as the context: must terminate the initialization
      ErrorMsg currentError = new ErrorMsg(3, 2, LocalizableMessage.raw("current error"));
      currentError.setCreationTime(startTime);
      broker3.publish(currentError);

      DirectoryException failure = task.waitForCompletion(30, TimeUnit.SECONDS);
      assertNotNull(failure, "the initialization completed without an error");
      assertEquals(failure.getMessageObject().toString(), "current error",
          "the ErrorMsg timestamped before the context must stay ignored");
      assertFalse(domain2.ieRunning(),
          "the terminated initialization must release the import/export context");
    }
    finally
    {
      stop(broker3);
      disable(domain2);
      remove(replServer);
    }
  }

  /**
   * When the initialization request receives no answer at all - the publish
   * was silently dropped or the answer was lost - the stalled-request watchdog
   * must fail the task after the configured delay instead of letting it wait
   * forever (issue #861).
   */
  @Test(enabled=true)
  public void stalledInitializeFromRemoteIsAborted() throws Exception
  {
    DN testService = DN.valueOf("o=test");
    ReplicationServer replServer = null;
    FakeReplicationDomain domain2 = null;
    ReplicationBroker broker3 = null;

    try
    {
      int replServerPort = TestCaseUtils.findFreePort();
      replServer = createReplicationServer(14, replServerPort,
          "stalledInitializeRequestDb", 100);
      SortedSet<String> servers = newTreeSet("localhost:" + replServerPort);

      domain2 = new FakeReplicationDomain(
          testService, 2, servers, 0, null, new StringBuffer(), 0);
      // broker3 receives the InitializeRequestMsg and never answers it
      broker3 = openReplicationSession(testService, 3, 100, replServerPort,
          10000, domain2.getGenerationID());

      waitTopologyKnowsReplica(domain2, 3);

      RecordingInitializeTask task = new RecordingInitializeTask();
      domain2.initializeFromRemote(3, task);

      assertFalse(domain2.abortStalledInitializeFromRemote(60000),
          "the initialization must not be aborted before the delay elapses");

      final FakeReplicationDomain requester = domain2;
      TestTimer abortTimer = new TestTimer.Builder()
          .maxSleep(30, SECONDS)
          .sleepTimes(10, MILLISECONDS)
          .toTimer();
      abortTimer.repeatUntilSuccess(() -> assertTrue(
          requester.abortStalledInitializeFromRemote(50),
          "the stalled initialization was never aborted"));

      DirectoryException failure = task.waitForCompletion(30, TimeUnit.SECONDS);
      assertNotNull(failure, "the stalled initialization must fail the task");
      assertEquals(failure.getMessageObject().toString(),
          ERR_NO_REACHABLE_PEER_IN_THE_DOMAIN.get(testService, 3).toString());
      assertFalse(domain2.ieRunning(),
          "the aborted initialization must release the import/export context");
      assertFalse(domain2.abortStalledInitializeFromRemote(0),
          "a second abort must be a no-op once the context is released");
    }
    finally
    {
      stop(broker3);
      disable(domain2);
      remove(replServer);
    }
  }

  private void waitTopologyKnowsReplica(ReplicationDomain domain, int dsId) throws Exception
  {
    TestTimer timer = new TestTimer.Builder()
        .maxSleep(30, SECONDS)
        .sleepTimes(100, MILLISECONDS)
        .toTimer();
    timer.repeatUntilSuccess(() -> assertTrue(domain.getReplicaInfos().containsKey(dsId),
        "DS(" + dsId + ") is not known to the domain"));
  }

  private String buildExportedData(final int ENTRYCOUNT)
  {
    final StringBuilder sb = new StringBuilder();