/*
|
* The contents of this file are subject to the terms of the Common Development and
|
* Distribution License (the License). You may not use this file except in compliance with the
|
* License.
|
*
|
* You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
|
* specific language governing permission and limitations under the License.
|
*
|
* When distributing Covered Software, include this CDDL Header Notice in each file and include
|
* the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
|
* Header, with the fields enclosed by brackets [] replaced by your own identifying
|
* information: "Portions Copyright [year] [name of copyright owner]".
|
*
|
* Copyright 2009 Sun Microsystems, Inc.
|
* Portions Copyright 2011-2015 ForgeRock AS.
|
* Portions Copyright 2026 3A Systems, LLC.
|
*/
|
package org.opends.server.replication.plugin;
|
|
import java.util.ArrayList;
|
import java.util.Collection;
|
import java.util.List;
|
import java.util.Map.Entry;
|
import java.util.NoSuchElementException;
|
import java.util.TreeMap;
|
import java.util.concurrent.Callable;
|
|
import org.forgerock.i18n.LocalizableMessage;
|
import org.forgerock.i18n.slf4j.LocalizedLogger;
|
import org.forgerock.util.annotations.VisibleForTesting;
|
import org.opends.server.replication.common.CSN;
|
import org.opends.server.replication.common.CSNGenerator;
|
import org.opends.server.replication.protocol.LDAPUpdateMsg;
|
import org.opends.server.replication.protocol.ReplicaOfflineMsg;
|
import org.opends.server.replication.protocol.UpdateMsg;
|
import org.opends.server.replication.service.ReplicationDomain;
|
import org.opends.server.types.operation.PluginOperation;
|
|
/**
|
* This class is used to store the list of local operations currently
|
* in progress and not yet committed in the database.
|
* <p>
|
* It is used to make sure that operations are sent to the Replication Server
|
* in the order defined by their CSN. It is also used to update the ServerState
|
* at the appropriate time.
|
* <p>
|
* An object of this class is instantiated for each ReplicationDomain.
|
*/
|
class PendingChanges
|
{
|
private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
|
|
/**
|
* How many local changes a recovery keeps at most, and how many bytes their messages take at
|
* most: past either, it keeps none, and searches for them as for the changes done before it began.
|
*/
|
static final int MAX_KEPT_CHANGES = 10000;
|
static final long MAX_KEPT_BYTES = 32L * 1024 * 1024;
|
|
/** A map used to store the pending changes. */
|
private final TreeMap<CSN, PendingChange> pendingChanges = new TreeMap<>();
|
|
/**
|
* The {@link CSNGenerator} to use to create new unique CSNs
|
* for each operation done on the replication domain.
|
*/
|
private final CSNGenerator csnGenerator;
|
|
/** The ReplicationDomain that will be used to send UpdateMsg. */
|
private final ReplicationDomain domain;
|
|
/** Told that the replica of this domain announces itself offline, or takes that back. */
|
private final ReplicaOfflineAnnouncer replicaOfflineAnnouncer;
|
|
/**
|
* Changed under the lock the domain starts and ends a recovery under - and ended under the lock of
|
* this list as well - and read without either by {@link #isRecovering()}.
|
*/
|
private volatile boolean recoveringOldChanges;
|
/** What the recovery which runs keeps of the local changes: {@code null} until it first needs it. */
|
private Recovery recovery;
|
/** How many local changes a recovery keeps at most: fewer in the tests which make it run over. */
|
private int maxKeptChanges = MAX_KEPT_CHANGES;
|
|
/**
|
* Creates a new PendingChanges using the provided CSNGenerator.
|
*
|
* @param csnGenerator The CSNGenerator to use to create new unique CSNs.
|
* @param domain The ReplicationDomain that will be used to send UpdateMsg.
|
* @param replicaOfflineAnnouncer Told that the replica of this domain announces itself
|
* offline, before the message announcing it is published, and that it takes
|
* the announcement back when the broker refused the message.
|
*/
|
PendingChanges(CSNGenerator csnGenerator, ReplicationDomain domain,
|
ReplicaOfflineAnnouncer replicaOfflineAnnouncer)
|
{
|
this.csnGenerator = csnGenerator;
|
this.domain = domain;
|
this.replicaOfflineAnnouncer = replicaOfflineAnnouncer;
|
}
|
|
/**
|
* Remove and return an update form the pending changes list.
|
*
|
* @param csn
|
* The CSN of the update to remove.
|
*/
|
synchronized void remove(CSN csn)
|
{
|
pendingChanges.remove(csn);
|
}
|
|
/**
|
* Returns the number of update currently in the list.
|
*
|
* @return The number of update currently in the list.
|
*/
|
int size()
|
{
|
return pendingChanges.size();
|
}
|
|
/**
|
* Mark an update message as committed.
|
*
|
* @param csn The CSN of the update message that must be set as committed.
|
* @param msg The message associated to the update.
|
*/
|
private synchronized void commit(CSN csn, LDAPUpdateMsg msg)
|
{
|
final PendingChange curChange = pendingChanges.get(csn);
|
if (curChange == null)
|
{
|
throw new NoSuchElementException();
|
}
|
curChange.setCommitted(true);
|
curChange.setMsg(msg);
|
}
|
|
/**
|
* Add a new UpdateMsg to the pending list from the provided local operation.
|
*
|
* @param operation The local operation for which an UpdateMsg must
|
* be added in the pending list.
|
* @return The CSN now associated to the operation.
|
*/
|
synchronized CSN putLocalOperation(PluginOperation operation)
|
{
|
final CSN csn = csnGenerator.newCSN();
|
if (!operation.isSynchronizationOperation())
|
{
|
pendingChanges.put(csn, new PendingChange(csn, operation, null));
|
}
|
return csn;
|
}
|
|
/**
|
* Add a replica offline message to the pending list and publish it, if the changes which
|
* come before it have all been published.
|
* <p>
|
* The message carries the newest CSN of the replica, so a change which is still in flight
|
* holds it back - and there is nobody left to publish it afterwards: the caller announces
|
* the replica offline while its service is being disabled, and the broker stops right after.
|
* Such a message is given up on rather than left queued, so that it is neither reported as
|
* sent nor published later on the session which follows.
|
*
|
* @return the CSN of the message which was published, or {@code null} if it could not be:
|
* a change which is still in flight holds it back, or the broker refused it
|
*/
|
public synchronized CSN putReplicaOfflineMsg()
|
{
|
final CSN offlineCSN = csnGenerator.newCSN();
|
final PendingChange pendingChange =
|
new PendingChange(offlineCSN, null, new ReplicaOfflineMsg(offlineCSN));
|
pendingChange.setCommitted(true);
|
|
pendingChanges.put(offlineCSN, pendingChange);
|
/*
|
* The message is the last change of the queue, so a push which did not reach it, or which
|
* the broker refused, reports another CSN or none.
|
*/
|
final boolean published = offlineCSN.equals(pushCommittedChanges());
|
// pushCommittedChanges() removes whatever it reached, so the message is still listed here
|
// if and only if a change before it held it back - it is dropped rather than left queued.
|
pendingChanges.remove(offlineCSN);
|
return published ? offlineCSN : null;
|
}
|
|
/**
|
* Push all committed local changes to the replicationServer service.
|
*
|
* @return the CSN of the last {@link ReplicaOfflineMsg} the replication service accepted, or
|
* {@code null} if none was pushed or the broker refused it. The announcement that a
|
* replica goes offline is the one message whose delivery the caller must know about:
|
* it is stored nowhere, so nothing publishes it again, while a change the broker
|
* refuses is republished from the historical information of its entry on the next
|
* session.
|
*/
|
synchronized CSN pushCommittedChanges()
|
{
|
CSN publishedOfflineCSN = null;
|
|
// peek the oldest change
|
Entry<CSN, PendingChange> firstEntry = pendingChanges.firstEntry();
|
if (firstEntry == null)
|
{
|
return null;
|
}
|
|
PendingChange firstChange = firstEntry.getValue();
|
|
while (firstChange != null && firstChange.isCommitted())
|
{
|
final PluginOperation op = firstChange.getOp();
|
final UpdateMsg msg = firstChange.getMsg();
|
if (msg instanceof LDAPUpdateMsg
|
&& op != null
|
&& !op.isSynchronizationOperation())
|
{
|
if (!recoveringOldChanges)
|
{
|
domain.publish(msg);
|
}
|
else
|
{
|
// do not push updates until the RS catches up: the recovery publishes them.
|
// Kept before the state takes it: the first change kept fixes where the search of the
|
// recovery ends, at the last change pushed before it.
|
// @see #setRecovering(boolean)
|
recovery().keep((LDAPUpdateMsg) msg);
|
domain.getServerState().update(msg.getCSN());
|
}
|
}
|
else if (msg instanceof ReplicaOfflineMsg && recoveringOldChanges)
|
{
|
/*
|
* The announcement carries the newest CSN of this replica, past the changes the recovery
|
* has still to publish: the replication server would take this replica for offline from
|
* there - its changelog index past the changes which reach it afterwards. It is given up
|
* on, as one a change in flight holds back is, whatever the broker would refuse.
|
*/
|
if (logger.isTraceEnabled())
|
{
|
logger.trace("The announcement that the replica goes offline at " + msg.getCSN()
|
+ " is given up on: a recovery has changes before it to publish");
|
}
|
}
|
else if (msg instanceof ReplicaOfflineMsg)
|
{
|
/*
|
* Announce the replica offline before the message reaches the wire, and not after:
|
* a collocated replication server forwards it as soon as it has it, and a forward
|
* which finds nothing announced leaves the shutdown waiting out the whole grace
|
* period of a message the topology already has.
|
*/
|
final CSN offlineCSN = msg.getCSN();
|
replicaOfflineAnnouncer.announce(offlineCSN);
|
if (domain.publish(msg))
|
{
|
publishedOfflineCSN = offlineCSN;
|
}
|
else
|
{
|
// The broker wrote it to no session, so nobody will forward what was announced.
|
replicaOfflineAnnouncer.withdraw(offlineCSN);
|
}
|
}
|
|
// false warning: firstEntry will not be null if firstChange is not null
|
pendingChanges.remove(firstEntry.getKey());
|
|
// peek the oldest change
|
firstEntry = pendingChanges.firstEntry();
|
firstChange = firstEntry != null ? firstEntry.getValue() : null;
|
}
|
return publishedOfflineCSN;
|
}
|
|
/**
|
* Mark an update message as committed, then
|
* push all committed local changes to the replicationServer service
|
* in a single atomic operation.
|
*
|
* @param csn The CSN of the update message that must be set as committed.
|
* @param msg The message associated to the update.
|
*/
|
synchronized void commitAndPushCommittedChanges(CSN csn, LDAPUpdateMsg msg)
|
{
|
commit(csn, msg);
|
pushCommittedChanges();
|
}
|
|
/**
|
* Set the PendingChangesList structure in a mode where it is waiting for the
|
* RS to receive all the previous changes to be sent before starting to
|
* process the changes normally. In this mode, The Domain does not publish the
|
* changes from the pendingChanges because there are older changes that need
|
* to be published before.
|
*
|
* <p>
|
* The local changes pushed meanwhile are kept for the recovery, which publishes them once it has
|
* published the ones done before it began. A recovery which is already running goes on, with what
|
* it kept: a session which starts while it runs starts no other.
|
* <p>
|
* It is set without the lock of this list, which the start of a session must not take, but under
|
* the one the domain ends a recovery under as well: a recovery the session needs is not ended by
|
* one which was about to end.
|
*
|
* @param recovering
|
* The recovering status that must be set.
|
*/
|
public void setRecovering(boolean recovering)
|
{
|
recoveringOldChanges = recovering;
|
}
|
|
/**
|
* Tells whether the local changes are left to a recovery, which publishes them.
|
* <p>
|
* It is read without the lock of this list, which the start of a session must not take: the
|
* session only learns from it that it needs a missing change publisher to end the recovery,
|
* and that one ends it under the lock.
|
*
|
* @return {@code true} if the local changes are left to a recovery
|
*/
|
boolean isRecovering()
|
{
|
return recoveringOldChanges;
|
}
|
|
/**
|
* Tells whether the recovery which runs keeps the local changes, and where its search of the
|
* historical information ends then.
|
*
|
* @return the recovery which keeps the local changes, or {@code null} if its search has to go up
|
* to the last local change: no recovery runs, or it kept too many changes to keep any
|
*/
|
synchronized KeptChanges keptChanges()
|
{
|
if (!recoveringOldChanges)
|
{
|
return null;
|
}
|
final Recovery running = recovery();
|
return running.kept != null ? new KeptChanges(running.searchUpTo) : null;
|
}
|
|
/**
|
* Returns the local changes the recovery which runs kept which come after the provided CSN, in
|
* order.
|
*
|
* @param csn
|
* where the recovery has got to
|
* @return the changes, or {@code null} if no recovery runs, or it kept too many changes to keep any
|
*/
|
synchronized List<LDAPUpdateMsg> keptChangesAfter(CSN csn)
|
{
|
final KeptChanges keeping = keptChanges();
|
return keeping != null ? new ArrayList<>(recovery.kept.tailMap(csn, false).values()) : null;
|
}
|
|
/**
|
* Returns the local changes the recovery which runs kept which come after the provided CSN, in
|
* order, whatever it searches for: the data the domain read its state from again may not reflect
|
* them.
|
*
|
* @param newest
|
* the newest change of this replica the data holds, or {@code null} if it holds none
|
* @return the changes, none if no recovery keeps any
|
*/
|
synchronized List<LDAPUpdateMsg> keptChangesNewerThan(CSN newest)
|
{
|
if (recovery == null || recovery.kept == null)
|
{
|
return new ArrayList<>();
|
}
|
return new ArrayList<>((newest != null ? recovery.kept.tailMap(newest, false) : recovery.kept).values());
|
}
|
|
/**
|
* Allows to update the recovery situation by comparing the CSN of the last
|
* change that was sent to the ReplicationServer with the CSN of the last
|
* operation that was taken out of the PendingChanges list. If the two match
|
* then the recovery is completed and normal procedure can restart. Otherwise
|
* the RSUpdater thread must continue to look for older changes and no changes
|
* can be committed from the pendingChanges list.
|
* <p>
|
* The recovery ends at once for every change pushed after it: a change which comes here once
|
* this list publishes again must find the broker taking it as well, or it is refused, while
|
* the domain state records it as done and no recovery is left to publish it. The recovery is
|
* therefore ended here, under the lock {@link #pushCommittedChanges()} takes, before any change
|
* can be pushed. What it kept is dropped with it.
|
*
|
* @param recovered
|
* The CSN of the last change that was published to the
|
* ReplicationServer.
|
* @param recoveryEnd
|
* Ends the recovery, unless a session which started meanwhile needs it, and has the
|
* broker take the changes again.
|
* @return A boolean indicating if the recovery is completed (false) or must
|
* continue (true).
|
*/
|
synchronized boolean recoveryUntil(CSN recovered, RecoveryEnd recoveryEnd)
|
{
|
final CSN lastLocalChange = domain.getLastLocalChange();
|
if (recoveringOldChanges && recovered != null && recovered.isNewerThanOrEqualTo(lastLocalChange))
|
{
|
recoveryEnd.end(() ->
|
{
|
recoveringOldChanges = false;
|
recovery = null;
|
});
|
}
|
return recoveringOldChanges;
|
}
|
|
/**
|
* Drops the provided local changes the recovery which runs kept: the data of the domain, read
|
* again, does not hold them - it was restored or imported since they were done - and the replicas
|
* would get changes this one does not hold.
|
*
|
* @param csns
|
* the CSNs of the changes to drop
|
*/
|
synchronized void dropKeptChanges(Collection<CSN> csns)
|
{
|
if (recovery == null || recovery.kept == null || csns.isEmpty())
|
{
|
return;
|
}
|
recovery.kept.keySet().removeAll(csns);
|
recovery.keptBytes = 0;
|
for (LDAPUpdateMsg msg : recovery.kept.values())
|
{
|
recovery.keptBytes += msg.size();
|
}
|
}
|
|
/**
|
* Has the recovery which runs keep no more local changes, and drop the ones it kept: it searches
|
* for them in the historical information instead, as one which kept too many does.
|
*/
|
synchronized void keepNoMoreChanges()
|
{
|
if (recovery != null)
|
{
|
recovery.kept = null;
|
}
|
}
|
|
/**
|
* Returns what the recovery which runs keeps of the local changes, under the lock of this list.
|
* It is made as it is first needed - as the recovery keeps its first change, or as its publisher
|
* asks where its search ends - so the search ends at the last change pushed before any was kept.
|
*/
|
private Recovery recovery()
|
{
|
if (recovery == null)
|
{
|
recovery = new Recovery(domain.getLastLocalChange());
|
}
|
return recovery;
|
}
|
|
/**
|
* Sets how many local changes a recovery which starts from now on keeps at most: with none, it
|
* searches for all of them in the historical information.
|
*
|
* @param maxKeptChanges
|
* how many local changes a recovery keeps at most
|
*/
|
@VisibleForTesting
|
synchronized void setMaxKeptChanges(int maxKeptChanges)
|
{
|
this.maxKeptChanges = maxKeptChanges;
|
}
|
|
/**
|
* What a recovery keeps of the local changes pushed while it runs.
|
* <p>
|
* A recovery republishes what the replication server misses from the historical information of
|
* the entries, and its search of it races the commits: a change committed while it searches is in
|
* the domain state, and not in what it found. The changes pushed once it began are kept instead,
|
* in order and with their messages, so that its search only has to cover the ones pushed before:
|
* up to the last of them, which no commit moves. A change kept is published as it was done, where
|
* a later change of the same values leaves nothing of it in the historical information, and a
|
* delete kept does not depend on the deletes the domain keeps aside, which are capped.
|
* <p>
|
* What it keeps stays until it ends: a missing change publisher whose broker refused a change,
|
* or whose session was replaced by one whose replication server holds less, leaves them to the
|
* one which goes on, which may start from further back. Past {@link #maxKeptChanges} changes or
|
* {@link #MAX_KEPT_BYTES} bytes it keeps none, and the recovery searches for them from where it
|
* has got to, holding the local changes back at the end, as it does when nothing is kept.
|
*/
|
private final class Recovery
|
{
|
/** The last local change pushed before this recovery kept any: its search ends there. */
|
private final CSN searchUpTo;
|
/** The local changes pushed since, in order; {@code null} once there were too many. */
|
private TreeMap<CSN, LDAPUpdateMsg> kept;
|
private long keptBytes;
|
|
private Recovery(CSN searchUpTo)
|
{
|
this.searchUpTo = searchUpTo;
|
this.kept = maxKeptChanges > 0 ? new TreeMap<CSN, LDAPUpdateMsg>() : null;
|
}
|
|
private void keep(LDAPUpdateMsg msg)
|
{
|
if (kept == null)
|
{
|
return;
|
}
|
/*
|
* Published as a recovery publishes what it rebuilds: not assured. Its operation may still be
|
* waiting for the acknowledgement when a round publishes it - nothing orders the two - and it
|
* then gets none and times out, as it does when the round comes later: it waits as the domain
|
* is configured to, on the CSN of the message, and not as the message says.
|
*/
|
if (msg.isAssured())
|
{
|
msg.setNotAssured();
|
}
|
kept.put(msg.getCSN(), msg);
|
keptBytes += msg.size();
|
if (kept.size() > maxKeptChanges || keptBytes > MAX_KEPT_BYTES)
|
{
|
kept = null;
|
if (logger.isDebugEnabled())
|
{
|
logger.debug(LocalizableMessage.raw("the recovery keeps no more local changes after " + msg.getCSN()
|
+ ": it searches for them in the historical information"));
|
}
|
}
|
}
|
}
|
|
/** A recovery which keeps the local changes pushed while it runs. */
|
static final class KeptChanges
|
{
|
/**
|
* The last local change pushed before the recovery kept any: its search ends there, and the
|
* changes it kept come next. {@code null} if the domain state held no change of this replica.
|
*/
|
final CSN searchUpTo;
|
|
private KeptChanges(CSN searchUpTo)
|
{
|
this.searchUpTo = searchUpTo;
|
}
|
}
|
|
/** Ends a recovery the pending changes have caught up with. */
|
interface RecoveryEnd
|
{
|
/**
|
* Ends the recovery, unless a session which started meanwhile has given it changes of its own
|
* to recover: runs {@code recovered} under the lock a recovery is started under, and has the
|
* broker take the changes again. Called under the lock of the pending changes.
|
*
|
* @param recovered
|
* lets the pending changes publish the local changes again
|
*/
|
void end(Runnable recovered);
|
}
|
|
/**
|
* Runs the provided rounds of a recovery while no local change can be numbered or pushed.
|
* <p>
|
* A round publishes the changes done when it began, and a change done while it runs is left
|
* to a round after it, so a recovery could chase the local changes for as long as they go on.
|
* The rounds run here are the last ones: what they publish is all there is, and nothing can be
|
* pushed between their end and the end of the recovery they decide. The local operations wait
|
* for them - for their CSN and for their push - for as long as publishing the changes done while
|
* the round before ran takes, where a publish in normal operation holds the same lock for one
|
* change.
|
*
|
* @param <T>
|
* what the rounds return
|
* @param round
|
* the rounds of the recovery
|
* @return what the round returned
|
* @throws Exception
|
* if the round failed
|
*/
|
synchronized <T> T holdBackLocalChangesWhile(Callable<T> round) throws Exception
|
{
|
return round.call();
|
}
|
|
/**
|
* Told that the replica of this domain announces itself offline, or takes that back.
|
* <p>
|
* A collocated replication server can forward a {@link ReplicaOfflineMsg} as soon as it is on
|
* the wire, and its shutdown waits for that forward, so the announcement has to be in place
|
* before the message is published: one made afterwards is one the forward found nothing to
|
* clear, and the shutdown spends its whole grace period on a message which has already gone
|
* out. The broker may still refuse the message once it is announced, and then the
|
* announcement is withdrawn: what stays announced is what the broker reports as written.
|
*/
|
interface ReplicaOfflineAnnouncer
|
{
|
/**
|
* Announces that the replica goes offline at the provided CSN.
|
*
|
* @param offlineCSN
|
* the CSN of the ReplicaOfflineMsg which is about to be published
|
*/
|
void announce(CSN offlineCSN);
|
|
/**
|
* Withdraws the announcement of a message the broker refused: it was written to no session,
|
* so nobody will forward it.
|
*
|
* @param offlineCSN
|
* the CSN of the ReplicaOfflineMsg which was announced and not published
|
*/
|
void withdraw(CSN offlineCSN);
|
}
|
}
|
