From c6650aedac5c1ade8aee9e7a76eb0dd100a3e179 Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Mon, 02 Nov 2026 15:39:56 +0000
Subject: [PATCH] [#1246] Keep the messages of the last local changes, and have a recovery publish the changes done before it began from them, as they were done (#1251)
---
opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java | 225 +++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 223 insertions(+), 2 deletions(-)
diff --git a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java
index 378cc37..ef2c4f7 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/replication/plugin/PendingChanges.java
@@ -19,9 +19,11 @@
import java.util.ArrayList;
import java.util.Collection;
+import java.util.HashSet;
import java.util.List;
import java.util.Map.Entry;
import java.util.NoSuchElementException;
+import java.util.Set;
import java.util.TreeMap;
import java.util.concurrent.Callable;
@@ -56,9 +58,33 @@
*/
static final int MAX_KEPT_CHANGES = 10000;
static final long MAX_KEPT_BYTES = 32L * 1024 * 1024;
+ /**
+ * How many of the last local changes the domain keeps the messages of at most, and how much of the
+ * heap those take at most, as {@link #footprint(LDAPUpdateMsg)} reckons it: a recovery publishes the
+ * changes its replication server misses from them, and rebuilds the ones before them from the
+ * historical information - see {@link LastLocalChanges}. Under the load of the replication convergence
+ * run, a few hundred changes a second, a replication server more than two minutes behind is covered.
+ * <p>
+ * Each domain written to keeps that much, whether a recovery runs or not, so its bytes are a sixteenth
+ * of the heap at most; multiple written domains can keep that share each. A server given a small heap
+ * covers a shorter lag. A recovery round holds the messages it publishes besides: while it runs, as much
+ * again can stay on the heap.
+ */
+ static final int MAX_RECENT_CHANGES = 100000;
+ static final long MAX_RECENT_BYTES = Math.min(64L * 1024 * 1024, Runtime.getRuntime().maxMemory() / 16);
+ /**
+ * What a message kept takes besides twice its {@link LDAPUpdateMsg#size()}, which counts its encoded
+ * changes once, and its encoding holds them again: its DN, entry UUID and CSN, the objects of its
+ * changes, and its place in the map. Measured on the messages of the load of the convergence run,
+ * kept as they are here: 880 bytes of heap for a modify whose size() is 168, 1300 for an add whose
+ * size() is 370.
+ */
+ static final int MESSAGE_OVERHEAD = 560;
/** A map used to store the pending changes. */
private final TreeMap<CSN, PendingChange> pendingChanges = new TreeMap<>();
+ /** Local operations pending at a reload, which may have written to the replaced data. */
+ private final Set<CSN> pendingAtDataReload = new HashSet<>();
/**
* The {@link CSNGenerator} to use to create new unique CSNs
@@ -81,6 +107,10 @@
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;
+ /** The messages of the last local changes, which a recovery publishes as they were done. */
+ private final LastLocalChanges lastLocalChanges = new LastLocalChanges();
+ /** How many of the last local changes the domain keeps the messages of at most: fewer in tests. */
+ private int maxRecentChanges = MAX_RECENT_CHANGES;
/**
* Creates a new PendingChanges using the provided CSNGenerator.
@@ -108,6 +138,7 @@
synchronized void remove(CSN csn)
{
pendingChanges.remove(csn);
+ pendingAtDataReload.remove(csn);
}
/**
@@ -193,8 +224,8 @@
* {@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.
+ * refuses is republished on the next session, from the message kept of it or, once that
+ * is dropped, from the historical information of its entry.
*/
synchronized CSN pushCommittedChanges()
{
@@ -217,6 +248,16 @@
&& op != null
&& !op.isSynchronizationOperation())
{
+ // Kept before the domain state takes it, published or not: a recovery publishes it as it was done.
+ if (pendingAtDataReload.contains(firstChange.getCSN()))
+ {
+ // There is a gap in the messages kept after the reload; earlier ones cannot cover it.
+ lastLocalChanges.forget();
+ }
+ else
+ {
+ lastLocalChanges.add(firstChange.getCSN(), (LDAPUpdateMsg) msg);
+ }
if (!recoveringOldChanges)
{
domain.publish(msg);
@@ -268,6 +309,7 @@
// false warning: firstEntry will not be null if firstChange is not null
pendingChanges.remove(firstEntry.getKey());
+ pendingAtDataReload.remove(firstChange.getCSN());
// peek the oldest change
firstEntry = pendingChanges.firstEntry();
@@ -536,6 +578,185 @@
}
}
+ /**
+ * Returns the messages the domain kept of its local changes which come after the provided CSN, up
+ * to the provided one, in order, or tells where the messages it kept begin if they do not go back
+ * that far.
+ *
+ * @param after
+ * where the recovery has got to
+ * @param upTo
+ * where the changes to publish end, or {@code null} for the last local change
+ * @return the messages, or where the ones kept begin
+ */
+ synchronized RecentChanges recentChangesAfter(CSN after, CSN upTo)
+ {
+ return lastLocalChanges.after(after, upTo != null ? upTo : domain.getLastLocalChange());
+ }
+
+ /**
+ * Forgets the messages kept of the last local changes: the domain read its state from data which
+ * may lack their changes - it was restored or imported since - and a recovery which published them
+ * would have the other replicas do changes this one does not hold. Local operations still pending
+ * may have written to the replaced data even if they have not yet committed, so their messages are
+ * not kept when they are pushed. Later messages are kept, and a recovery rebuilds the changes before
+ * them from the historical information.
+ */
+ synchronized void forgetRecentChanges()
+ {
+ lastLocalChanges.forget();
+ for (PendingChange change : pendingChanges.values())
+ {
+ if (change.getOp() != null && !change.getOp().isSynchronizationOperation())
+ {
+ pendingAtDataReload.add(change.getCSN());
+ }
+ }
+ }
+
+ /**
+ * Sets how many of the last local changes the domain keeps the messages of at most: with none, a
+ * recovery rebuilds every change from the historical information.
+ *
+ * @param maxRecentChanges
+ * how many of the last local changes the domain keeps the messages of at most
+ */
+ @VisibleForTesting
+ synchronized void setMaxRecentChanges(int maxRecentChanges)
+ {
+ this.maxRecentChanges = maxRecentChanges;
+ lastLocalChanges.trim();
+ }
+
+ /**
+ * The messages of the last local changes, in order, which a recovery publishes as they were done.
+ * <p>
+ * A recovery publishes the changes its replication server misses of this replica, and rebuilt from
+ * the historical information of the entries, those are not the changes: it holds nothing of a change
+ * whose values a later one replaced, nor of the add of an entry deleted since. A replication server
+ * which took the later changes takes none of the older ones afterwards - from the peer which got them
+ * from this replica either, whose changes it takes in order, as a log ordered by CSN - so a directory
+ * server which failed over to a replication server behind it, or which the load balancing moved to
+ * one, left gaps in its changelog for good (issue #1246). The message of every local change pushed -
+ * published, refused by the broker, or left to a recovery - is kept here instead, for as many changes
+ * as {@link #maxRecentChanges} and {@link #MAX_RECENT_BYTES} allow, the oldest dropped first.
+ * <p>
+ * Every local change after {@link #keptAfter} has its message kept: the CSN right before the first one
+ * kept, and the newest one dropped once some were. A recovery from further back rebuilds the changes up
+ * to there from the historical information. It is not the last local change the domain state holds:
+ * the state read back once the data was restored, or once the domain was enabled again, can be older
+ * than changes done since the domain started - a delete leaves no historical information to bring it
+ * up to - and taking the messages kept for the changes after it skipped those changes.
+ */
+ private final class LastLocalChanges
+ {
+ private final TreeMap<CSN, LDAPUpdateMsg> messages = new TreeMap<>();
+ private long bytes;
+ /**
+ * Every local change after this one has its message kept; {@code null} if none was kept since the
+ * domain started or forgot them.
+ */
+ private CSN keptAfter;
+
+ private void add(CSN csn, LDAPUpdateMsg msg)
+ {
+ if (maxRecentChanges <= 0)
+ {
+ return;
+ }
+ if (keptAfter == null)
+ {
+ keptAfter = justBefore(csn);
+ }
+ // A change is pushed again when the publish of the push before threw: it is counted once.
+ final LDAPUpdateMsg replaced = messages.put(csn, msg);
+ bytes += footprint(msg) - (replaced != null ? footprint(replaced) : 0);
+ trim();
+ }
+
+ private void trim()
+ {
+ if (maxRecentChanges <= 0)
+ {
+ forget();
+ return;
+ }
+ while (!messages.isEmpty() && (messages.size() > maxRecentChanges || bytes > MAX_RECENT_BYTES))
+ {
+ final Entry<CSN, LDAPUpdateMsg> oldest = messages.pollFirstEntry();
+ bytes -= footprint(oldest.getValue());
+ keptAfter = oldest.getKey();
+ }
+ }
+
+ private void forget()
+ {
+ messages.clear();
+ bytes = 0;
+ keptAfter = null;
+ }
+
+ private RecentChanges after(CSN after, CSN upTo)
+ {
+ if (keptAfter == null)
+ {
+ return new RecentChanges(null, upTo, null);
+ }
+ if (after.isOlderThan(keptAfter))
+ {
+ return new RecentChanges(null, upTo, keptAfter);
+ }
+ final List<LDAPUpdateMsg> found = upTo != null && after.isOlderThan(upTo)
+ ? new ArrayList<>(messages.subMap(after, false, upTo, true).values())
+ : new ArrayList<LDAPUpdateMsg>();
+ return new RecentChanges(found, upTo, keptAfter);
+ }
+ }
+
+ /**
+ * Returns the CSN right before the provided one among the CSNs of its server: no CSN of that server
+ * comes in between. For a sequence number of 0, it is the greatest one of the millisecond before, as
+ * {@code Integer.MAX_VALUE} rather than -1: a search window of a recovery may start there, and the
+ * index of the historical information reads -1 as the greatest sequence number of its millisecond.
+ */
+ private static CSN justBefore(CSN csn)
+ {
+ return csn.getSeqnum() > 0
+ ? new CSN(csn.getTime(), csn.getSeqnum() - 1, csn.getServerId())
+ : new CSN(csn.getTime() - 1, Integer.MAX_VALUE, csn.getServerId());
+ }
+
+ /** Returns how much of the heap a message kept takes, as reckoned for {@link #MAX_RECENT_BYTES}. */
+ static long footprint(LDAPUpdateMsg msg)
+ {
+ return 2L * msg.size() + MESSAGE_OVERHEAD;
+ }
+
+ /** What the domain kept of the messages of its local changes from a CSN on, up to another one. */
+ static final class RecentChanges
+ {
+ /**
+ * The messages of the local changes after the CSN asked from, up to {@link #upTo}, in order;
+ * {@code null} if the messages kept do not go back that far, or the domain keeps none.
+ */
+ final List<LDAPUpdateMsg> messages;
+ /** Where the changes asked for end: the CSN asked for, or the last local change. */
+ final CSN upTo;
+ /**
+ * Every local change after this one has its message kept: where the changes to rebuild from the
+ * historical information end when {@link #messages} is {@code null}. {@code null} if no message is
+ * kept, and the changes are all to rebuild.
+ */
+ final CSN keptAfter;
+
+ private RecentChanges(List<LDAPUpdateMsg> messages, CSN upTo, CSN keptAfter)
+ {
+ this.messages = messages;
+ this.upTo = upTo;
+ this.keptAfter = keptAfter;
+ }
+ }
+
/** A recovery which keeps the local changes pushed while it runs. */
static final class KeptChanges
{
--
Gitblit v1.10.0