[#1041] Claim the import context for the length of a session restart, and refuse the total update which lands across it

[vharseko]

The owner read of a session restart and the listener thread's claim of the import context share no lock, and cannot share the one the restart holds: disableService() waits for the listener thread under serviceStateLock, so a listener which took that lock before its claim would wait for a restart which is waiting for it. A restart which read no owner a few statements before the claim landed stopped the broker the import was about to read - and then waited for the listener thread to run that import to its end over the stopped broker:

• the stop landing before the first entry, which is where it lands - between the read and shutdown = true there is one socket write, and the listener has preBackendImport() to get through before it reads anything - the import ends on nothing, importLDIF replaces the suffix with the nothing which arrived, and loadDataState() finds no base entry and computes the generationId of an empty suffix. It was reported as a finished total update with no error until Replication: a total update whose session stops before the DoneMsg ends as a finished import, with the exporter's generationId over partial data #1039 landed ([#1039] Report a total update whose session stops before the DoneMsg as a failed import #1044); on this master it is recorded as a failed import (ERR_INIT_SESSION_STOPPED_DURING_IMPORT) and the replica no longer comes up as a peer of the exporter over the nothing it holds - the data of this replica is gone either way;
• the stop landing before setNewStatus(TO_FULL_UPDATE_STATUS_EVENT), the listener thread dies on a NullPointerException in signalStatusChange() - connectedRS.get().session is null once the broker is stopped - after initialize()'s finally has reported the total update as finished. This is what the first run of the new test showed; the issue describes the first shape.

What this changes

• The two contend on the one reference instead of one reading what the other writes. ReplicationDomain.disableServiceUnlessImportInProgress() claims importExportContext with a context of its own, SESSION_BEING_STOPPED, before it stops anything, and releases it once disableService() has returned - the listener thread is gone by then, and it is the one thread which claims a total update this replica did not ask for. restartSession() calls it in place of sessionHasAnOwner() + disableService(): a claim which fails against an import leaves the session alone, as the read did; one which fails against an export stops the session from under it, as before - an export is not an owner, and the exporter reports the cut.
• The listener's claim fails against SESSION_BEING_STOPPED, and the InitializeTargetMsg is refused: ERR_INIT_REJECTED_SESSION_STOPPING (330) is sent to the exporter and logged on this server, which had nothing in its log for a total update it refused. The text names the refusing server by its id: the exporter shows only the details of the ErrorMsg, with no sender, so "this directory server" would have named the exporter there. A refusal published over a broker which is already stopped is silently not sent, and the exporter then learns of it from the replication server, as it did.
• Once its claim is made, the listener reads again whether the broker is stopping, and refuses the total update if it is. That narrows what the claim does not reach: the session stopped by disable() or by the shutdown of the domain, neither of which claims, and the total update which lands between the end of an export and the stop of a restart which found that export. Without it the claim wins, and what runs next publishes the full update status over a session which is gone. It is a read, not a claim - none of those roads writes anything it could fail against - so a stop which lands after it still has the import run over a session which is going down, and end as a failed import over the suffix it has replaced (Replication: a total update whose session stops before the DoneMsg ends as a finished import, with the exporter's generationId over partial data #1039). Every one of these roads had that window before this change, and has it still.
• The answer to a total update this replica asked for itself (initiator == getServerId()) gets the same read once markInitStartReceived() has taken the context the request claimed: a restart never stops the session under it - an import owns it - but disable() and the shutdown of the domain do, and nothing released that context on those roads. The task which asked for the total update is failed with 330. Pre-existing, and the same window as the arm above.
• The claim is invisible to ieRunning(), importInProgress(), getImportExportContext() and every reader of the reference: the ErrorMsg and InitializeRcvAckMsg arms of receive(), the self-initiated branch of initialize() - markInitStartReceived() on the claim would return true and run the import on a static context - abortStalledInitializeFromRemote(), the export, and the monitor.
• releaseIEContext() takes the context to release and releases only that one. initializeFromRemote() released the context on every failure, including the failure to acquire one: a dsreplication initialize run while an import or export was in progress released that operation's context - receiveEntryBytes() then reads a null context - and, with this change, would have released the claim of a restart. Pre-existing, and on the road this change makes the claim depend on.

Tests

Two seams, after setReplayDrainTimeout(): setImportClaimHook() runs on the listener thread between the InitializeTargetMsg it took off the session and its claim - or, for a total update this replica asked for, its read of the context the request claimed - setServiceStopHook() inside disableService() before the broker is stopped. Nothing else can hold either thread there.

Four cases in ReplayDuringImportTest, on the fixture of #968 (the exporter is a broker of the test, the replay queue is synchronous):

• aRestartDecidedBeforeTheImportIsClaimedRefusesTheImport - the listener held before its claim; a Modify whose entryUUID search never runs (ShortCircuitPlugin on SEARCH) spends its attempts in place on a thread of its own, decides the restart and is held before the stop; the listener is released in between, so the broker it finds is up and only the claim can refuse it. The refusal reaches the exporter as an ErrorMsg with that text, NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START is not logged, the listener thread does not die, the session comes back after the backoff, and the entry added before is still there. While the stop is held, the exporter also sends an answer to a request of this replica which no context stands for (the request abandoned as stalled, dsreplication enable hangs initializing registration information when adding a third replica; CI Test replication step has no timeout #861), followed by a total update of its own as the point the listener is known to be past it: the answer is ignored rather than imported into the claim (that arm reading the raw reference in place of getImportExportContext() starts the total update, and is red). Without the claim - sessionHasAnOwner() read and disableService() in its place - nothing refuses the total update asked for while the stop is held, and the case is red there.
• aDomainDisabledBeforeTheImportIsClaimedRefusesTheImport - the same hold, disable() across it, enable() after. With the re-read of the broker after the claim removed it fails on the start of the total update.
• aDomainDisabledBeforeTheImportItAskedForStartsRefusesTheImport - the same, for a total update this replica asked for: initializeFromRemote(), the answer held before the listener reads the context, disable() across it. With the read of the broker removed from that arm the import starts over the stopped session, and the case is red.
• aRestartDecidedWhileAnExportRunsStopsTheSessionItStreamsOver - the export arm. The domain exports into the exporter broker, which never reports the start of the total update, so the export stands in waitForRemoteStartOfInit() holding the context; the same failed replay decides the restart, and the restart is held in the stop seam. The stop is reached - an export is not an owner - the export ends on the cut it reports, and the session comes back. Under if (owner != null) in place of if (owner.importInProgress()) the restart returns false, the stop seam is never reached and the case is red.

What each of the four roads out of the claim is pinned by, one assertion each: the sentinel is invisible to ieRunning() while the stop is held (getImportExportContext() returning the raw reference is red); a total update asked for while it is held is refused and does not release it (initializeFromRemote() back to the unconditional importExportContext.set(null) is red); the sentinel is released with the stop, so the next total update into this replica runs to its end (deleting the CAS in the finally is red); and the import the re-read refuses gives its context back (deleting that releaseIEContext() is red). The guard and the CAS of the refused-request road are red only together: with the guard in place the refused road releases nothing, so the CAS in releaseIEContext() on its own - which keeps a second completion of the same context (the ErrorMsg arm and the stalled watchdog each hold their own read of it) from clearing what was claimed in between - is not pinned; no case drives two completions.

ReplayDuringImportTest is 10/10 on this head, and the twelve other classes which drive a total update or read the import/export context - SessionRestartBackoffTest, SessionRestartRequestsTest, ServerStateFlushTest, NamingConflictTest, PendingChangesTest, ReplicationDomainTest, InitOnLineTest, GenerationIdTest, ReSyncTest, UpdateOperationTest, AssuredReplicationPluginTest and MonitorTest - are 131/131, Skipped: 0.

On master

The claim replaces the read #968 introduced (sessionHasAnOwner()) and the refusal is the road that read left open a few statements wide. The branch is the one commit [#1041] on origin/master at 26defc28ef, rebased across #977 ([#952]), #981, #982, #1044 ([#1039]), #1042 ([#1040]), and now #1010 ([#915]), #1019 ([#1014]) and #1022 ([#1021]).

The one conflict of this rebase was again the tail of replication.properties: #1019 added WARN_IGNORING_UPDATE_UNSUPPORTED_BY_PEER_328 where this branch adds 330, and both blocks are kept, master's first. The Java sources merged on their own.

With #1049

#1049 (#1048) has runRequestedSessionRestarts() - the only replay-road caller of restartSession() - return at once while ieRunning(), so a restart waits out an export instead of cutting it. With both merged that is the contract: the export arm of disableServiceUnlessImportInProgress() is then reached only across the window between that read and the claim, and aRestartDecidedWhileAnExportRunsStopsTheSessionItStreamsOver - which holds the restart in the stop seam under an export - goes red, although the two changes touch different methods and rebase cleanly. The PR merged second drops that case and narrows the javadoc of disableServiceUnlessImportInProgress() to that window.

Ordinal

ERR_INIT_REJECTED_SESSION_STOPPING_330: master holds up to 329 (with #1044 and #1019 merged), and #1049 holds 331 among the open PRs. 330 is free on master and on every open PR.

Not in this change

• Replication: the session restart a configuration change asks for cuts a total update into this replica #1040 - restartService(), the restart a configuration change asks for, is merged as [#1040] Leave the session to a total update when a configuration change asks for a restart #1042 and still decides on sessionHasAnOwner(); the claim is the primitive it would take.
• Replication: a total update whose session stops before the DoneMsg ends as a finished import, with the exporter's generationId over partial data #1039 - merged as [#1039] Report a total update whose session stops before the DoneMsg as a failed import #1044: a stop which lands once the import is reading its entries no longer ends it as a finished import over partial data, it records the failure and computes the generationId over the data. The data is replaced all the same, and the claim is what keeps the replay's restart from being that stop; the re-read narrows the roads which make no claim, and the shutdown of the server mid-import is still one of them.
• signalStatusChange() on a stopped broker throws a NullPointerException out of setNewStatus() and ends the listener thread on it. This change keeps the total update from reaching it over a stopped broker; the null session itself is pre-existing and deserves an issue of its own.

Fixes #1041

[vharseko]

@maximthomas rebased on origin/master now that #968 is merged: the branch is the one commit [#1041] (8287944616), the [#956] commits are gone from it.

Two things changed on the way, both in replication.properties only:

• the conflict was the tail of the file - [#952] Keep a failed state write from killing the checkpointer and hanging the shutdown #977 added 323-324 there after [#956] Tell a failed entryUUID search apart from an entry which is not there #968 - and is resolved by keeping both blocks;
• the ordinal moves from 329 to 330: [#1039] Report a total update whose session stops before the DoneMsg as a failed import #1044 had taken 329 with a different key twelve minutes before this PR was opened, so the "329 is free" claim in the description was wrong from the start. The Java constant carries no number and the tests read .ordinal(), so nothing else changes.

git range-diff against the previous head shows only the context of that hunk; the tree is master's plus this commit (git diff old head → new head is exactly master's delta after #968). test-compile of opendj-server-legacy is clean with the generated constant at 330. #977 touches nothing the claim reads or writes - its ieRunning() sees through the claim like every other reader - and the description now says so in place of the "Built on #968" section.

[vharseko]

@maximthomas rebased onto master at c6b483127f - the branch was conflicting with #981 - and the one [#1041] commit goes over it with one conflict, in ReplayDuringImportTest: #981 appended aRequestWhichStoodWhileTheImportRanIsNotRunOnceItIsOver at the point where this PR appends its two cases. All three are kept, master's first.

LDAPReplicationDomain merged on its own, in restartSession(), and the merge holds: #981 reads wakes = sessionRestartBackoffWakes() under the lock right after the lines where this PR replaces sessionHasAnOwner() + disableService() with the claim, and the early return is what it was - a claim which fails against an import leaves the session alone. What #981 changes around the claim is who comes to it: the restart a released change asks for is a request now, run by the thread which made it or by the state checkpointer when no thread comes back for it, and both roads come to restartSession(), so the claim guards the checkpointer's restart the way it guards the replay thread's.

ReplayDuringImportTest 6/6, SessionRestartBackoffTest 3/3 on this head, Skipped: 0. The description says so in "On master", and its ordinal section says master holds 325.

[vharseko]

@maximthomas rebased onto master at d07bb31897, now that #982 has landed - the branch was conflicting - and the one [#1041] commit goes over it (eccfe94d93). Nothing new is proposed here; this only says what the conflict was and how it was taken.

The conflict was in ReplayDuringImportTest, in two places, both adjacency: #982 appends aWarningAfterTheImportDoesNotCountTheDeliveriesBefore at the point where this PR appends aRestartDecidedBeforeTheImportIsClaimedRefusesTheImport and aDomainDisabledBeforeTheImportIsClaimedRefusesTheImport - master's case is kept ahead of them - and it adds a paragraph to the class javadoc where this PR adds its own; both are kept. LDAPReplicationDomain, ReplicationDomain and replication.properties merged on their own: #982 is in the retry warning and in the resets disable() and the end of a total update share, and the claim reads none of that. git range-diff against 972bd1bb78 shows the claim and the refusal byte for byte the same.

Re-run after the rebase: ReplayDuringImportTest 7/7 - the case of #982 among them - and SessionRestartBackoffTest 3/3, Skipped: 0. The description says so.

[maximthomas]

praise: The claim sits where the race is, and the release fix beside it closes a leak older than this PR.

• ReplicationDomain.disableServiceUnlessImportInProgress() (:3502) claims the reference under serviceStateLock, the one lock restartSession() already holds, and releases it in a finally once the listener thread is joined — the one thread which claims a total update this replica did not ask for.
• releaseIEContext(ImportExportContext) (:2089) as a CAS on the given context, with initializeFromRemote() releasing only what it acquired (:2517-2520): at the base a dsreplication initialize refused against a live import or export wiped that operation's context.
• getImportExportContext() (:4022-4025) as the single reader keeps the sentinel out of ieRunning(), importInProgress() and the monitor, so the checkpointer goes on saving across a restart.

---

issue (non-blocking): The post-claim re-read narrows the stop roads which make no claim; the description and the javadoc say it closes them.

opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java:2639-2650, :3496-3498, :3514-3519
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/LDAPReplicationDomain.java:5683-5684
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationBroker.java:2722-2745

if (broker.shuttingDown()) at :2639 is a check-then-act: nothing orders it against broker.stop() on the roads which call disableService() bare — disable() (LRD :5050), shutdown() (LRD :2668), restartService() (:3659), the config roads (LRD :1031, :4157), and the export arm of disableServiceUnlessImportInProgress() itself (:3514-3519: the CAS lost to an export, the sentinel never set). The listener cannot take serviceStateLock (the join inside disableService() would wait for it). A stop landing after the re-read runs the total update over a stopped broker: the signalStatusChange() NPE if before setNewStatus, else importLDIF over an empty stream — the first shape of #1041. The window is not a few statements: importBackend() runs importingData = true; awaitReplayDrained(); (LRD :5683-5684, bounded by replayDrainTimeoutInMs) and preBackendImport() before the first broker.receive() (:2160-2171). On the export arm the restart is still before shutdown = true — broker.stop() publishes the ReplicaOfflineMsg first (RB :2722-2745) — when the export releases and the listener claims and reads false. The body says "That closes what the claim does not reach" and "the re-read keeps a disable() or a shutdown from being it before the first entry"; the javadoc at :3496-3498 says the export-then-stop total update "is refused by the listener on the broker it finds stopping". Both hold only when shutdown = true precedes the re-read. Every one of these roads had the same window at the base, so this is not a regression; if the residual is left to #1039 deliberately, the fix is the wording alone.

   * not an owner: the session is stopped from under it and the exporter reports the cut, as
   * it does for every other stop. A total update which lands between the end of that export
   * and the stop is refused by the listener when it reads the broker as stopping after its
   * claim; a stop which lands after that read and before the first entry is received still
   * ends it as an import over nothing (issue #1039) - the read narrows that window, it does
   * not close it.

Or: close the export arm the way the sentinel closes the no-owner arm — a second flag with each side writing before it reads (the restart sets sessionStopping before it reads the context, the listener reads sessionStopping after its CAS; the worst case is a spurious refusal). disable()/shutdown() stay #1039's: closing them means deciding what a stop does to a running import.

---

issue (non-blocking): ERR_INIT_REJECTED_SESSION_STOPPING_330 describes an import into this server, and is thrown for an export out of it as well.

opendj-server-legacy/src/messages/org/opends/messages/replication.properties:704
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java:2072-2074, :1645

acquireIEContext(boolean importInProgress) picks the text on importExportContext.get() == SESSION_BEING_STOPPED alone; initializeRemote() acquires through the same CAS at :1645 with false. An export started while a restart holds the sentinel — an InitializeRequestMsg taken off the session just before the stop, or a local InitializeTargetTask for the length of disableService() — is logged at :1667 and reported to the requester or in the task result as "The total update of domain "%s" into this directory server was refused: ... the entries would have been read over that session". The refusal is right; nothing branches on the text; the task result describes an import nobody asked for.

      final LocalizableMessage message = importExportContext.get() != SESSION_BEING_STOPPED
          ? ERR_SIMULTANEOUS_IMPORT_EXPORT_REJECTED.get()
          : importInProgress
              ? ERR_INIT_REJECTED_SESSION_STOPPING.get(getBaseDN())
              : ERR_EXPORT_REJECTED_SESSION_STOPPING.get(getBaseDN());

Or: one direction-neutral text — "The total update of domain "%s" was refused by this directory server: its session to the replication server is being stopped, and the entries would have been streamed over that session."

---

suggestion (non-blocking): The "release only your own context" fix has no red test.

opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java:2089, :2517-2520
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java:493

No test calls initializeFromRemote() against an owner or against the claim: ReplayDuringImportTest:268 and ReplicationDomainTest acquire on a free context, InitOnLineTest.initializeSimultaneous is enabled = false, ReplicationDomainTest:520-541 refuses initializeRemote, whose catch never released. Each single mutant is equivalent by construction (with the guard every caller passes its own context; without it compareAndSet(null, null) is a no-op); the double mutant — the catch back to an unconditional importExportContext.set(null) — restores the base's leak and every case stays green.

      assertTrue(domain.isConnected(), "the session was stopped before the stop was held");
      assertThatThrownBy(() -> domain.initializeFromRemote(EXPORTER_ID, null))
          .as("a total update asked for while the session is being stopped was not refused")
          .isInstanceOf(DirectoryException.class)
          .hasMessageContaining(ERR_INIT_REJECTED_SESSION_STOPPING.get(baseDN).toString());

Pin: in aRestartDecidedBeforeTheImportIsClaimedRefusesTheImport, while the stop is held; the case then goes on as it stands — under the double mutant the refused request wipes the claim, the listener's CAS wins and the NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START count at :503 goes red.

---

suggestion (non-blocking): The export arm of disableServiceUnlessImportInProgress() is unpinned.

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

The arm is observable only as a replay-driven restart under an export out of this replica. The classes with an export (GenerationIdTest, InitOnLineTest, FractionalReplicationTest, ProtocolCompatibilityTest, SynchronizationMsgTest, ReplicationDomainTest) and the classes with a failed-replay restart fixture (UpdateOperationTest, AssuredReplicationPluginTest, IsServerFailureTest, NamingConflictTest, ReplayDuringImportTest, SessionRestart*Test) are disjoint; the mutant if (owner.importInProgress()) -> if (owner != null) — any owner leaves the session alone — is green on every case.

// on the ReplayDuringImportTest fixture: the domain exports into the exporter broker, which
// reads nothing, so the export holds the context; then a replay fails the way test 1 does
final AtomicReference<Throwable> exportFailure = new AtomicReference<>();
final Thread export = new Thread(() -> {
  try { domain.initializeRemote(EXPORTER_ID, null); } catch (Throwable t) { exportFailure.set(t); }
}, "export");
export.start();
waitUntil(domain::ieRunning, "the export did not claim the context");
// ShortCircuitPlugin on SEARCH + replayMsg(...) as in aRestartDecidedBeforeTheImportIsClaimedRefusesTheImport
waitUntil(() -> !domain.isConnected(), "the restart left the session to the export");
waitUntil(domain::isConnected, "the session was not started back after the restart");

Pin: under the mutant the restart returns false and the session never drops — the first waitUntil goes red.

---

suggestion (non-blocking): The release of a freshly claimed import on the re-read arm is unpinned.

opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java:2648
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java:593

aDomainDisabledBeforeTheImportIsClaimedRefusesTheImport asserts the entry, the START count, the listener deaths and the refusal count; none reads the context after enable(). Delete releaseIEContext(ieCtx); at :2648 and the import context stays claimed for the domain's life — ieRunning() true, the checkpointer skips every ds-sync-state save (LRD :723) silently, every later total update refused as simultaneous — and the case is green.

    waitUntil(domain::isConnected, "the session was not started back by enable()");
    assertFalse(domain.ieRunning(), "the refused import left its context claimed");

---

suggestion (non-blocking): The release of SESSION_BEING_STOPPED once the stop is done is unpinned.

opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java:3527
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java:525-530

Nothing in aRestartDecidedBeforeTheImportIsClaimedRefusesTheImport claims a total update after the restart. Delete the finally's CAS at :3527 and the sentinel stays: getImportExportContext() hides it from every reader, a later restart's loop reads importInProgress() false and stops as before, and every later acquireIEContext — a dsreplication initialize into this replica, a remote InitializeTargetMsg, an export — is refused with 330 until the domain object goes. The case is green.

    waitUntil(domain::isConnected, "the session was not started back after the restart");
    // The claim of the stop was released: the next total update into this replica runs.
    startImportInto(exportedEntries().length);
    finishImport(exportedEntries());
    assertThat(errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()))
        .as("the total update after the restart did not run to its end")
        .hasSize(totalUpdatesEndedBefore + 2);

Pin: under the mutant the second total update is refused with 330 and startImportInto times out on the backend never being deregistered. The existing hasSize(totalUpdatesEndedBefore) at :530 moves before the second import.

---

suggestion (non-blocking): The sentinel filter in getImportExportContext() is unpinned.

opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java:4022-4025
opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java:493

The sentinel is held only inside restartSession(); test 1 holds the stop at :486-493 and asserts only domain.isConnected() there, test 2 holds nothing across a claim, and finishImport polls ieRunning() after a millisecond stop is over. The mutant return ieCtx; — the claim visible to ieRunning(), the checkpointer's save at LRD :723 and the self-initiated arm of initialize() — is green on every case.

      assertTrue(domain.isConnected(), "the session was stopped before the stop was held");
      assertFalse(domain.ieRunning(), "the claim of the stop is visible as a running import");

Pin: the hook runs after the CAS at :3506, inside disableService(); the line is red under the mutant.

---

nitpick (non-blocking): The refusal is logged at ERROR; the sibling refusal of a remote's export request is logged at NOTICE.

opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java:2567, :1667

rejectInitializeTarget() does logger.error(reason) on both arms — including the re-read after a routine disable() or a server shutdown, and the pre-existing simultaneous-import arm, which logged nothing locally at the base — while :1667, this server refusing a remote's export request on the same detail, is logger.info(NOTE_FULL_UPDATE_REMOTE_REQUEST_REJECTED). An operator who disables a domain with an InitializeTargetMsg in flight gets an ERROR line on the replica which did nothing wrong. The tests match msgID= in the error log and count NOTE records through the same helper, so logger.info with a NOTE-category message keeps them green. Your call if the ERROR is wanted.

[vharseko]

@maximthomas round 2 on 7f9adeddf2, rebased onto master at b960e40edc. Two of the things this review stands on moved with the base while it was being written: #1039 landed as #1044 and #1040 as #1042.

---

issue: the post-claim re-read narrows the stop roads which make no claim; the description and the javadoc say it closes them.

Taken, as the wording. if (broker.shuttingDown()) is a read, and none of the roads which call disableService() bare writes anything the claim could fail against, so nothing orders the two - and the window is importingData = true; awaitReplayDrained(); plus preBackendImport() wide, not a few statements. The javadoc of disableServiceUnlessImportInProgress(), the comment at the re-read and the description now say that, with the residual as it stands on this master: since #1044 a stop which lands after the read no longer ends the import as a finished one over nothing - the failure is recorded and the generationId is computed over the data - but the suffix is replaced all the same, so the read narrows the window and does not close it.

   * not an owner: the session is stopped from under it and the exporter reports the cut, as
   * it does for every other stop. A total update which lands between the end of that export
   * and the stop is refused by the listener when it reads the broker as stopping after its
   * claim; a stop which lands after that read still has the import run over a session which
   * is going down, and end as a failed import over the suffix it has replaced (issue
   * #1039): the read narrows that window, it does not close it.

The second flag is not taken. It would close the export arm the way the sentinel closes the no-owner arm and leave disable(), the shutdown and the configuration roads where they are: one road out of five, for a second piece of state on the same contract, with the same worst case as the read (a total update refused although the stop never came). What the export arm is worth pinning as is that the session is stopped under an export, and round 2 pins that.

---

issue: ERR_INIT_REJECTED_SESSION_STOPPING_330 describes an import into this server, and is thrown for an export out of it as well.

Taken, as the one direction-neutral text. initializeRemote() acquires through the same CAS with false, and what an export gets back is reported in the task result or to the remote which asked for it - as an import nobody asked for, as you say.

ERR_INIT_REJECTED_SESSION_STOPPING_330=The total update of domain "%s" was refused by this \
 directory server: its session to the replication server is being stopped, and the entries \
 would have been streamed over that session. Ask for the total update again once the session \
 is back

One text rather than two: nothing branches on it, the key is direction-neutral already, and an ordinal in a file the open PRs are appending to is worth not taking twice. The comment at acquireIEContext() says which directions the refusal covers.

---

suggestion: the "release only your own context" fix has no red test.

Taken, where you put it - in aRestartDecidedBeforeTheImportIsClaimedRefusesTheImport, while the stop is held.

      assertThatThrownBy(() -> domain.initializeFromRemote(EXPORTER_ID, null))
          .as("a total update asked for while the session is being stopped was not refused")
          .isInstanceOf(DirectoryException.class)
          .hasMessageContaining(ERR_INIT_REJECTED_SESSION_STOPPING.get(baseDN).toString());

Under the double mutant - the CAS back to importExportContext.set(null) and the null guard in initializeFromRemote() gone - the refused request wipes the claim, the listener's CAS wins when it is released and the case is red on the start count, exactly as you said: [a total update claimed against a restart which was decided was started] Expected size: 0 but was: 2.

It also moved where the no-claim mutant is caught: with sessionHasAnOwner() + disableService() back in restartSession() nothing holds the context, the request is not refused and the case is now red on this line rather than on the start of the total update.

---

suggestion: the export arm of disableServiceUnlessImportInProgress() is unpinned.

Taken, as a case of its own: aRestartDecidedWhileAnExportRunsStopsTheSessionItStreamsOver. Your fixture - the domain exports into the exporter broker, which reports no start of a total update, so the export stands in waitForRemoteStartOfInit() holding the context - with the stop seam in place of the first waitUntil: what the case reads is then the decision the export was found by, rather than a session which is down for the length of a backoff.

      assertTrue(stopHeld.await(30, TimeUnit.SECONDS),
          "the restart left the session to the export: an export is not the owner a total"
              + " update into this replica is");

Under if (owner != null) the restart returns false, disableService() is never called and the case is red there.

---

suggestion: the release of a freshly claimed import on the re-read arm is unpinned.

Taken, the line as you wrote it, in aDomainDisabledBeforeTheImportIsClaimedRefusesTheImport.

    waitUntil(domain::isConnected, "the session was not started back by enable()");
    assertFalse(domain.ieRunning(), "the refused import left its context claimed");

---

suggestion: the release of SESSION_BEING_STOPPED once the stop is done is unpinned.

Taken, as the second total update at the end of case 1, with the existing count moved ahead of it.

    startImportInto(exportedEntries().length);
    finishImport(exportedEntries());
    assertThat(errorLogRecordsOf(NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_END.ordinal()))
        .as("the total update after the restart did not run to its end")
        .hasSize(totalUpdatesEndedBefore + 2);

With the finally's CAS deleted it is red on startImportInto timing out on the backend never being deregistered, as you said - and so is aWarningAfterTheImportDoesNotCountTheDeliveriesBefore of #982, which restarts the session and then imports too.

---

suggestion: the sentinel filter in getImportExportContext() is unpinned.

Taken, the line as you wrote it, where the stop is held and the claim is the only thing in the reference.

      assertFalse(domain.ieRunning(), "the claim of the stop is visible as a running import");

---

nitpick: the refusal is logged at ERROR; the sibling refusal of a remote's export request is logged at NOTICE.

Kept at ERROR, and this is the one thing I am not taking on its own terms - say the word and it goes to logger.info with a NOTE key.

The two are not the same event. :1667 is this server declining to start an export it was asked for: nothing of the requester's is lost, it is told, and the road is routinely raced by the topology propagation the comment there describes. The refusal here ends a total update an administrator asked for, over the session its entries were to be streamed over, and this server is the only place the reason exists - the exporter learns that the target refused, not why. An operator whose dsreplication initialize failed greps this replica's log for it.

The disable() and shutdown roads are that same event seen from the other end: the domain was disabled, or the server stopped, with an InitializeTargetMsg in flight, and the total update is dead for it. ERR_SIMULTANEOUS_IMPORT_EXPORT_REJECTED on the arm above is the gap this shares - a total update refused with no trace on the replica which refused it.

---

Runs on this head, class = JVM:

ReplayDuringImportTest 9/9, and the twelve other classes which drive a total update or read the import/export context - SessionRestartBackoffTest, SessionRestartRequestsTest, ServerStateFlushTest, NamingConflictTest, PendingChangesTest, ReplicationDomainTest, InitOnLineTest, GenerationIdTest, ReSyncTest, UpdateOperationTest, AssuredReplicationPluginTest and MonitorTest - 131/131, Skipped: 0.

Each road out of the claim, one mutant each, all on ReplayDuringImportTest:

mutant	red on
restartSession() back to sessionHasAnOwner() + disableService()	case 1, on the refusal of the total update asked for while the stop is held
the re-read after the claim deleted	case 2, on the start of the total update
releaseIEContext() back to importExportContext.set(null), with the null guard in initializeFromRemote() gone	case 1, on the start of the total update
getImportExportContext() returning the reference as it is	case 1, on ieRunning() while the stop is held
the releaseIEContext() of the re-read arm deleted	case 2, on ieRunning() after enable()
the CAS of the finally deleted	case 1 and the case of #982, both on startImportInto()
if (owner != null) in place of if (owner.importInProgress())	case 3, on the stop seam

[maximthomas]

praise: Round 2 turns every round-1 gift into a line that fails, and the new export case checks the decision itself, not a side effect of it.

• aRestartDecidedBeforeTheImportIsClaimedRefusesTheImport now asserts, while the stop is held, assertFalse(domain.ieRunning()) (:607) and that initializeFromRemote() is refused with 330 (:612-616). The claim is visible, and the failed acquire does not release it.
• aRestartDecidedWhileAnExportRunsStopsTheSessionItStreamsOver waits on the stop seam (:822), so the case reads the owner.importInProgress() decision in disableServiceUnlessImportInProgress() directly.
• The javadoc and the comment at the re-read now say it narrows the non-claiming stop roads and does not close them. That is accurate on a base that has #1044.

---

suggestion (non-blocking): The post-claim re-read of the broker is only on the remote-initiated arm of initialize(). A total update this replica asked for itself still runs over a session that is stopping.

opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java:2600-2618, :2644

The self-initiated arm (initiator == getServerId()) takes the context initializeFromRemote() acquired, calls markInitStartReceived() and goes straight into importBackend(), with no broker.shuttingDown() read. Nothing releases that pending context on the disable() or domain-shutdown roads: disableService() only stops the broker and joins the listener. If the stop lands across the answering InitializeTargetMsg, the import runs over the stopped broker and ends as ERR 329 over the suffix it has replaced (#1039). The remote arm now refuses that same interleaving at :2644. This gap was already there at the base, and the restart road cannot reach it (importInProgress() is true, so the session is left alone). The fix is either the same read, or one sentence in the javadoc saying this arm is left as it was.

      if (ieCtx == null || !ieCtx.markInitStartReceived())
      {
        ...
        return;
      }
      if (broker.shuttingDown())
      {
        ieCtx.setExceptionIfNoneSet(new DirectoryException(ResultCode.OTHER,
            ERR_INIT_REJECTED_SESSION_STOPPING.get(getBaseDN())));
        completeInitializeTask(ieCtx);
        return;
      }

---

question (non-blocking): Once #1049 is in, which contract stands: does a restart cut an export (this PR), or wait for it (#1049)?

opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java:822

This PR pins that a replay restart reaches the stop under an export ("an export is not an owner"). #1049 (head 2506e2a8) makes runRequestedSessionRestarts() start with if (ieRunning()) { return false; }, and that method is the only replay-road caller of restartSession(). With both merged, the restart waits out the export, the stop seam never runs, and stopHeld.await(30, TimeUnit.SECONDS) fails. The two changes touch different methods, so whichever one is rebased second will rebase cleanly and still go red. This does not block this PR, because its contract is consistent at this head. The PR merged second has to drop or invert this case, or narrow the hold to imports. Not built or run on the combined head.

---

suggestion (non-blocking): The new refused-request pin is red only when both defences are reverted. The CAS in releaseIEContext() is not pinned on its own.

opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java:607-616
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java:2090-2093, :2139, :2557

If the if (ieCtx != null) guard in initializeFromRemote() stays, the refused road never calls releaseIEContext(), so a body put back to importExportContext.set(null) keeps the case green. In round 1 I called that single mutant equivalent. It is not: completeInitializeTask() is reached from the ErrorMsg arm (:2139) and from the stalled watchdog (:2557), and each holds its own read of the same context. Under set(null), the second release clears whatever was claimed in between (the sentinel, or a new import). Under the CAS it does nothing. No case drives two completions. Pinning it needs a seam between the watchdog's read and its release. The cheaper option is to make the comment claim only what the case pins:

      /*
       * A total update asked for here is refused against the claim of the stop, and the
       * claim is left where it is: the road which fails to acquire a context of its own
       * releases nothing.
       */

---

suggestion (non-blocking): The readers moved to getImportExportContext() are pinned only through ieRunning(). The self-initiated arm of initialize() could read the sentinel again and no test would notice.

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

assertFalse(domain.ieRunning()) at :607 pins the getter, not the call sites that use it. If :2602 goes back to importExportContext.get(), it gets SESSION_BEING_STOPPED. markInitStartReceived() returns true on it (abandonedAsStalled is false), and an import then runs on the static sentinel that every domain shares. This is reachable when the #861 watchdog has abandoned the request and a restart claims before the stale InitializeTargetMsg arrives. The only case that holds the claim sends a non-self InitializeTargetMsg, so every case stays green. Not run.

Pin: while the stop is held in aRestartDecidedBeforeTheImportIsClaimedRefusesTheImport, have the exporter publish an InitializeTargetMsg whose initiator is DS_ID, and assert that the NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START count stays at totalUpdatesStartedBefore. It is red under ieCtx = importExportContext.get() at :2602.

---

nitpick (non-blocking): The javadoc of aRestartDecidedBeforeTheImportIsClaimedRefusesTheImport still describes the outcome from before #1039.

opendj-server-legacy/src/test/java/org/opends/server/replication/plugin/ReplayDuringImportTest.java:530-532, :651-653

It says the import "ends on nothing with no exception recorded … and the total update is reported as finished". The base contains #1044, and both the case's own comment at :651-653 and the commit message say "as a failed import since issue #1039".

   * thread about to run the import. Run over that broker, the import ends on the nothing
   * which arrived - as a failed import since issue #1039, and as a finished one before it -
   * over a suffix which has been replaced by it all the same. Here the listener is held

---

nitpick (non-blocking): ERR_INIT_REJECTED_SESSION_STOPPING_330 says "refused by this directory server", but the exporter reads it too.

opendj-server-legacy/src/messages/org/opends/messages/replication.properties:712-715
opendj-server-legacy/src/main/java/org/opends/server/replication/service/ReplicationDomain.java:2572-2575, :2121-2127

The same message is logged locally and sent as the ErrorMsg detail. The exporter shows only getDetails(), in the task result via processErrorMsg() or through ERR_ERROR_MSG_RECEIVED_79, with no sender id. On the exporter, "this directory server … its session" therefore names the exporter and the exporter's session. This wording was my own round-1 suggestion, and this corrects it: name the refusing server, as 329 does.

ERR_INIT_REJECTED_SESSION_STOPPING_330=The total update of domain "%s" was refused by directory \
 server %d: its session to the replication server is being stopped, and the entries would have \
 been streamed over that session. Ask for the total update again once the session is back

[vharseko]

Thank you for round 3. All six points are taken; the head is 7418d7bbf6, one commit [#1041] on origin/master 26defc28ef. The rebase had one conflict, the tail of replication.properties again: #1019 added 328 next to 330, and both blocks are kept, master's first.

suggestion - the self-initiated arm of initialize() has no read of the broker. Taken, as the read rather than the sentence. Once markInitStartReceived() has taken the context the request claimed, the arm reads broker.shuttingDown() and, if the session is stopping, fails the task with 330 through completeInitializeTask() - your snippet, with no ErrorMsg: the requester is this server, and the exporter learns of the stop the way it does of any other. The comment says it is the same read as the remote arm, with the same window. To pin it, importClaimHook now runs before the fork, so it holds this arm too (the javadoc of the field and of setImportClaimHook() say so). The new case aDomainDisabledBeforeTheImportItAskedForStartsRefusesTheImport calls initializeFromRemote(), holds the answer, and disables the domain across it. With the read removed from that arm the import starts over the stopped session, and the case is red.

question - #1049. Once both are in, #1049's contract stands: a replay restart waits out an export. runRequestedSessionRestarts() is the only replay-road caller of restartSession(), so the export arm of disableServiceUnlessImportInProgress() is reached only across the window between #1049's ieRunning() read and the claim, the window #1049's javadoc describes. aRestartDecidedWhileAnExportRunsStopsTheSessionItStreamsOver then holds nothing reachable and goes red on stopHeld.await. Whichever PR is merged second drops that case and narrows the javadoc of disableServiceUnlessImportInProgress() ("An export in the context is not an owner ...") to that window. The PR description of this one now says so under "With #1049". Nothing changes here at this head, as you note.

suggestion - the CAS in releaseIEContext() is not pinned on its own. Taken, the cheaper way: the comment now claims only what the case pins, in your words ("the road which fails to acquire a context of its own releases nothing"). Thanks for correcting round 1. You are right that the single mutant is not equivalent: the ErrorMsg arm and the stalled watchdog each complete their own read of the context. The CAS stays, and the javadoc of releaseIEContext() gives its reason. No seam was added for a double completion. The PR description records that the guard and the CAS are pinned only together.

suggestion - the readers moved to getImportExportContext() are pinned only through ieRunning(). Taken, with your pin and one change. While the stop is held in aRestartDecidedBeforeTheImportIsClaimedRefusesTheImport, the exporter publishes an InitializeTargetMsg whose initiator is DS_ID. It then publishes a second one with its own initiator, as a point the listener is known to be past the first: that one is refused against the claim and logs 330. The case waits for that refusal, or for a start, and asserts that NOTE_FULL_UPDATE_ENGAGED_FROM_REMOTE_START stays at totalUpdatesStartedBefore. Under ieCtx = importExportContext.get() at that arm the total update starts, and the case is red.

nitpick - the javadoc of aRestartDecidedBeforeTheImportIsClaimedRefusesTheImport. Taken with your wording: "as a failed import since issue #1039, and as a finished one before it".

nitpick - 330 names "this directory server". Taken: refused by directory server %d, with getServerId() at every site that builds it (acquireIEContext() and both arms of initialize()), and DS_ID in the test assertions.

Checks on this head:

• ReplayDuringImportTest: 10/10, both through Maven (-Pprecommit verify) and run directly.
• The twelve other classes from round 2: 131/131, Skipped: 0.
• Nine mutants, each red on its own case: the seven of round 2, plus the two of this round. With no read on the self-initiated arm, the new disable case is red. With the raw get() on that arm, the restart case is red on the stale answer.
• The javadoc gate is green.

[maximthomas]

praise: This round closes the round-2 findings in the code, not just in the text.

• The answer to a total update this replica asked for now gets the same post-claim broker.shuttingDown() read as a remote-initiated one, and fails the task with 330 (ReplicationDomain.java:2625-2638).
• ERR_INIT_REJECTED_SESSION_STOPPING_330 now names the refusing server, getServerId(), at all three callers (ReplicationDomain.java:2077, :2635, :2677). The exporter can see who refused.
• ReplayDuringImportTest passes 10/10 at 7418d7bb, including the new aDomainDisabledBeforeTheImportItAskedForStartsRefusesTheImport (local failsafe run, 128 s suite).