Skip to content

[#1055] Report a ReplicaOfflineMsg forwarded once it is written to the peer, not once it is queued #1057

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
vharseko merged 2 commits into from
Sep 25, 2026
Merged

[#1055] Report a ReplicaOfflineMsg forwarded once it is written to the peer, not once it is queued #1057

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6ae52a5
[#1055] Report a ReplicaOfflineMsg forwarded once it is written to th…
vharseko Sep 16, 2026
7c7f055
[#1055] Pin the take-back of a message queued after the close drained…
vharseko Sep 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
136 changes: 116 additions & 20 deletions opendj-server-legacy/src/main/java/org/opends/server/replication/protocol/Session.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.forgerock.util.annotations.VisibleForTesting;
import org.opends.server.api.DirectoryThread;
import org.opends.server.types.HostPort;
import org.opends.server.util.StaticUtils;
Expand Down Expand Up @@ -112,8 +113,27 @@ public final class Session extends DirectoryThread implements Closeable
*/
private BufferedOutputStream output;

private final LinkedBlockingQueue<byte[]> sendQueue = new LinkedBlockingQueue<>(4000);
/** A message queued for the thread of this session, and what to run once it is written. */
private static final class Outgoing
{
private final byte[] buffer;
private final Runnable whenWritten;

private Outgoing(byte[] buffer, Runnable whenWritten)
{
this.buffer = buffer;
this.whenWritten = whenWritten;
}
}

private final LinkedBlockingQueue<Outgoing> sendQueue = new LinkedBlockingQueue<>(4000);
private AtomicBoolean isRunning = new AtomicBoolean(false);
/**
* What {@link #publish(ReplicationMsg, Runnable)} runs between its check that no close has
* begun and the offer of the message to {@code sendQueue}, or null. Only the tests set it - see
* {@link #beforeQueueing(Runnable)}.
*/
private volatile Runnable beforeQueueing;
private final CountDownLatch latch = new CountDownLatch(1);

/**
Expand Down Expand Up @@ -166,8 +186,10 @@ public Session(final Socket socket,
* This object won't be used anymore after this method is called.
* <p>
* A message which was published on this session but which its publisher thread had not sent yet
* is sent here rather than dropped, within the budget of {@link #DRAIN_BUDGET_MS}. See {@link
* #sendWhatThePublisherLeftQueued()}.
* is sent here rather than dropped, within the budget of {@link #DRAIN_BUDGET_MS}, and its
* callback runs here once it is written. See {@link #sendWhatThePublisherLeftQueued()}. What the
* close gives up on is not written, and the callbacks of those messages never run - see
* {@link #publish(ReplicationMsg, Runnable)}.
*/
@Override
public void close()
Expand Down Expand Up @@ -341,8 +363,8 @@ public void close()
private void sendWhatThePublisherLeftQueued()
{
final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(DRAIN_BUDGET_MS);
byte[] buffer;
while ((buffer = sendQueue.poll()) != null)
Outgoing outgoing;
while ((outgoing = sendQueue.poll()) != null)
{
if (System.nanoTime() - deadline >= 0)
{
Expand All @@ -352,7 +374,7 @@ private void sendWhatThePublisherLeftQueued()
}
try
{
send(buffer);
send(outgoing.buffer);
}
catch (final IOException e)
{
Expand All @@ -367,6 +389,7 @@ private void sendWhatThePublisherLeftQueued()
"the write failed with " + e.getClass().getName() + ": " + e.getMessage());
return;
}
written(outgoing.whenWritten);
}
}

Expand Down Expand Up @@ -495,27 +518,58 @@ public boolean isEncrypted()
* If an IO error occurred.
*/
public void publish(final ReplicationMsg msg) throws IOException
{
publish(msg, null);
}

/**
* Sends a replication message to the remote peer, and runs the provided callback once the
* message has been written to the socket.
* <p>
* While the thread of this session runs, a message published is queued for it and written
* later, so the return of this method says only that the message is queued. The callback is
* the only word that the message has left this server: it runs once, on the thread which wrote
* the message, after the write returned - the thread of the session, or the one closing it for
* a message the close sends out of the queue - and never for a message which was not written,
* which is what becomes of a message the write of which fails, and of what a close gives up on
* (see {@link #close()}). It must be short and must not block: the session writes nothing else
* until it returns.
*
* @param msg
* The message to be sent.
* @param whenWritten
* What to run once the message has been written, or null.
* @return whether the message was written or queued to be written; false when it was neither,
* because it has no encoding for the protocol version of the peer or because the
* session is being closed - the callback then never runs. A message queued after a
* close drained the queue is taken back, and counts as neither.
* @throws IOException
* If an IO error occurred.
*/
public boolean publish(final ReplicationMsg msg, final Runnable whenWritten) throws IOException
{
final byte[] buffer = msg.getBytes(protocolVersion);
if (buffer == null)
{
// skip anything that cannot be encoded for this peer.
return;
return false;
}
if (isRunning.get())
{
final Outgoing outgoing = new Outgoing(buffer, whenWritten);
while (!closeInitiated)
{
final Runnable hook = beforeQueueing;
if (hook != null)
{
hook.run();
}
try
{
// Avoid blocking forever so that we can check for session closure.
if (sendQueue.offer(buffer, 100, TimeUnit.MILLISECONDS))
if (sendQueue.offer(outgoing, 100, TimeUnit.MILLISECONDS))
{
if (!isRunning.get())
{
takeBackWhatWasQueuedTooLate(buffer);
}
return;
return isRunning.get() || !takeBackWhatWasQueuedTooLate(outgoing);
}
}
catch (final InterruptedException e)
Expand All @@ -524,10 +578,27 @@ public void publish(final ReplicationMsg msg) throws IOException
throw new IOException(e.getMessage());
}
}
return false;
}
else
send(buffer);
written(whenWritten);
return true;
}

/** Runs what was to run once a message is written; a callback which fails takes nothing down. */
private void written(final Runnable whenWritten)
{
if (whenWritten != null)
{
send(buffer);
try
{
whenWritten.run();
}
catch (final RuntimeException e)
{
logger.error(LocalizableMessage.raw("The callback of a message written to %s failed: %s",
readableRemoteAddress, stackTraceToSingleLineString(e)));
}
}
}

Expand All @@ -542,23 +613,46 @@ public void publish(final ReplicationMsg msg) throws IOException
* what nothing sends. A buffer queued before the session came off the queueing branch is left
* to the drain - it cannot have seen the flag cleared - and a buffer the drain or the close
* already took is not found here, so nothing is reported twice.
*
* @return whether the message was taken back - it is then never written, and its callback never
* runs
*/
private void takeBackWhatWasQueuedTooLate(final byte[] buffer)
private boolean takeBackWhatWasQueuedTooLate(final Outgoing outgoing)
{
publishLock.lock();
try
{
if (sendQueue.remove(buffer))
if (sendQueue.remove(outgoing))
{
reportQueueNotSent(1, "it was queued after the publisher of the session had stopped");
return true;
}
return false;
}
finally
{
publishLock.unlock();
}
}

/**
* Sets what {@link #publish(ReplicationMsg, Runnable)} runs between its check that no close has
* begun and the offer of the message to the queue.
* <p>
* Only there for the tests of {@link #takeBackWhatWasQueuedTooLate(Outgoing)}: a
* {@code publish()} descheduled at that spot is the only one which can queue a message after a
* close has drained the queue, and nothing else holds a thread there on cue while the close
* runs to its end.
*
* @param hook
* What to run there, on the publishing thread, or null for nothing.
*/
@VisibleForTesting
void beforeQueueing(final Runnable hook)
{
beforeQueueing = hook;
}

/** Sends a replication message already encoded to the socket.
*
* @param buffer
Expand Down Expand Up @@ -769,25 +863,27 @@ public void run()
boolean needClosing = false;
while (!closeInitiated)
{
byte[] buffer;
Outgoing outgoing;
try
{
buffer = sendQueue.take();
outgoing = sendQueue.take();
}
catch (InterruptedException ie)
{
break;
}
try
{
send(buffer);
send(outgoing.buffer);
}
catch (IOException e)
{
setSessionError(e);
publisherFailedWrites.incrementAndGet();
needClosing = true;
continue;
}
written(outgoing.whenWritten);
}
/*
* A close clears the flag itself, under publishLock, once it has joined this thread - see
Expand Down
54 changes: 40 additions & 14 deletions opendj-server-legacy/src/main/java/org/opends/server/replication/server/ServerWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,9 +17,11 @@
*/
package org.opends.server.replication.server;

import java.io.IOException;
import java.net.SocketException;

import org.forgerock.i18n.LocalizableMessage;
import org.forgerock.opendj.ldap.DN;
import org.opends.server.api.DirectoryThread;
import org.forgerock.i18n.slf4j.LocalizedLogger;
import org.opends.server.replication.common.ServerStatus;
Expand Down Expand Up @@ -120,24 +122,14 @@ public void run()
replicationServerDomain.getBaseDN(), handler.getServerId());
}
}
else if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer())
{
forwardReplicaOfflineMsg((ReplicaOfflineMsg) updateMsg);
}
else
{
// Publish the update to the remote server using a protocol version it supports
session.publish(updateMsg);
/*
* Only the forward to a peer RS ends the wait of the shutdown: what the grace period
* buys is the rest of the topology learning that the replica went offline. A directory
* server is never handed this message - ReplicationServerDomain.put() does not queue
* it for one, and DataServerHandler.updateServerState() drops the one the changelog
* cursor of a directory server which is catching up synthesizes from the offline CSN
* of the replica (issue #1029) - so the guard says whose forward counts rather than
* telling two deliveries apart.
*/
if (updateMsg instanceof ReplicaOfflineMsg && !handler.isDataServer())
{
dsrsShutdownSync.replicaOfflineMsgForwarded(
replicationServerDomain.getBaseDN(), updateMsg.getCSN(), handler.getServerId());
}
}
}
}
Expand Down Expand Up @@ -170,6 +162,40 @@ public void run()
}
}

/**
* Publishes a ReplicaOfflineMsg to the peer replication server, and reports the forward to the
* shutdown which may be waiting for it.
* <p>
* Only the forward to a peer RS ends the wait of the shutdown: what the grace period buys is
* the rest of the topology learning that the replica went offline. A directory server is never
* handed this message - ReplicationServerDomain.put() does not queue it for one, and
* DataServerHandler.updateServerState() drops the one the changelog cursor of a directory
* server which is catching up synthesizes from the offline CSN of the replica (issue #1029) -
* so the guard of the caller says whose forward counts rather than telling two deliveries apart.
* <p>
* The forward is reported once the message has been written to the peer, not once it is queued
* for the thread of the session: the shutdown closes the session as soon as its wait ends, and
* Session.close() sends what is still queued only once the write it joins has returned, and
* only within a budget of its own, so a message reported forwarded while it was queued behind
* one the peer had not read yet would end the wait for a peer which had not been told, and
* leave its delivery to that budget rather than to the grace period. A message the session
* refuses - one published while the session is being closed - will never be written, and the
* shutdown must not wait for it. One the protocol version of the peer cannot carry is refused
* by the session as well, but does not get this far: isUpdateMsgFiltered() drops it and says so
* first (issue #1014).
*/
private void forwardReplicaOfflineMsg(final ReplicaOfflineMsg msg) throws IOException
{
final DN baseDN = replicationServerDomain.getBaseDN();
final int serverId = handler.getServerId();
final boolean accepted = session.publish(msg,
() -> dsrsShutdownSync.replicaOfflineMsgForwarded(baseDN, msg.getCSN(), serverId));
if (!accepted)
{
dsrsShutdownSync.replicaOfflineMsgNotForwarded(baseDN, serverId);
}
}

private boolean isUpdateMsgFiltered(UpdateMsg updateMsg)
{
if (!updateMsg.isEncodableFor(handler.getProtocolVersion()))
Expand Down
Loading
Loading