From 016502d22e0490ed8b1b4951f5e12be01c4a224c Mon Sep 17 00:00:00 2001
From: Valery Kharseko <vharseko@3a-systems.ru>
Date: Mon, 03 Aug 2026 16:13:03 +0000
Subject: [PATCH] [#807] Do not drop persistent search notifications through search-phase dedup (#812)

---
 opendj-server-legacy/src/main/java/org/opends/server/backends/ChangelogBackend.java |   90 ++++++++++++++++++++++++++++++++++++++------
 1 files changed, 77 insertions(+), 13 deletions(-)

diff --git a/opendj-server-legacy/src/main/java/org/opends/server/backends/ChangelogBackend.java b/opendj-server-legacy/src/main/java/org/opends/server/backends/ChangelogBackend.java
index 249f7b0..b899eb3 100644
--- a/opendj-server-legacy/src/main/java/org/opends/server/backends/ChangelogBackend.java
+++ b/opendj-server-legacy/src/main/java/org/opends/server/backends/ChangelogBackend.java
@@ -35,6 +35,7 @@
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Queue;
 import java.util.Set;
 import java.util.TimeZone;
 import java.util.concurrent.ConcurrentLinkedQueue;
@@ -399,7 +400,10 @@
       {
         final SearchOperation searchOp = pSearch.getSearchOperation();
         final CookieEntrySender entrySender = searchOp.getAttachment(ENTRY_SENDER_ATTACHMENT);
-        entrySender.persistentSearchSendEntry(baseDN, updateMsg);
+        if (!entrySender.persistentSearchSendEntry(baseDN, updateMsg))
+        {
+          stopPersistentSearch(pSearch);
+        }
       }
     }
     catch (DirectoryException e)
@@ -447,7 +451,10 @@
       {
         final SearchOperation searchOp = pSearch.getSearchOperation();
         final ChangeNumberEntrySender entrySender = searchOp.getAttachment(ENTRY_SENDER_ATTACHMENT);
-        entrySender.persistentSearchSendEntry(changeNumber, changeNumberEntry);
+        if (!entrySender.persistentSearchSendEntry(changeNumber, changeNumberEntry))
+        {
+          stopPersistentSearch(pSearch);
+        }
       }
     }
     catch (DirectoryException e)
@@ -875,14 +882,20 @@
   {
     initializePersistentSearch(pSearch);
 
-    if (isCookieBased(pSearch.getSearchOperation()))
+    final Queue<PersistentSearch> psearches = isCookieBased(pSearch.getSearchOperation())
+        ? cookieBasedPersistentSearches
+        : changeNumberBasedPersistentSearches;
+    psearches.add(pSearch);
+    // Without this, a cancelled persistent search keeps being handed the changes it can no longer
+    // report, for as long as this backend lives.
+    pSearch.registerCancellationCallback(new PersistentSearch.CancellationCallback()
     {
-      cookieBasedPersistentSearches.add(pSearch);
-    }
-    else
-    {
-      changeNumberBasedPersistentSearches.add(pSearch);
-    }
+      @Override
+      public void persistentSearchCancelled(PersistentSearch psearch)
+      {
+        psearches.remove(psearch);
+      }
+    });
     super.registerPersistentSearch(pSearch);
   }
 
@@ -1400,6 +1413,47 @@
     return true;
   }
 
+  /**
+   * Sends a change reported by the "persistent search" phase, if it matches the base, scope and
+   * filter of the current search operation. Contrary to the "initial search" phase, the change goes
+   * through the persistent search path: it is not bound by the size and time limits of the search,
+   * which are only lifted once the initial phase is over, and a change published in the meantime
+   * would be dropped without the client ever hearing about it.
+   *
+   * @return {@code true} if the persistent search should keep reporting changes, {@code false}
+   *         otherwise
+   */
+  private static boolean sendNotificationIfMatches(SearchOperation searchOp, Entry entry, String cookie)
+      throws DirectoryException
+  {
+    if (matchBaseAndScopeAndFilter(searchOp, entry))
+    {
+      return searchOp.returnPersistentSearchEntry(entry, getControls(cookie));
+    }
+    // maybe the next entry will match?
+    return true;
+  }
+
+  /**
+   * Stops the provided persistent search and tells the client, which would otherwise wait forever
+   * for changes on a search which no longer reports any.
+   */
+  private static void stopPersistentSearch(PersistentSearch pSearch)
+  {
+    try
+    {
+      // Before the cancellation, which deregisters this persistent search from the connection: the
+      // search operation left the operations in progress when its initial phase ended, so nothing
+      // would be left to hang the response on afterwards.
+      pSearch.getSearchOperation().sendSearchResultDone();
+    }
+    catch (Exception e)
+    {
+      logger.traceException(e);
+    }
+    pSearch.cancel();
+  }
+
   /** Indicates if the provided entry matches the filter, base and scope. */
   private static boolean matchBaseAndScopeAndFilter(SearchOperation searchOp, Entry entry) throws DirectoryException
   {
@@ -1647,12 +1701,17 @@
       return sendEntryIfMatches(searchOp, entry, null);
     }
 
-    private void persistentSearchSendEntry(long changeNumber, Entry entry) throws DirectoryException
+    /**
+     * @return {@code true} if the persistent search should keep reporting changes, {@code false}
+     *         otherwise
+     */
+    private boolean persistentSearchSendEntry(long changeNumber, Entry entry) throws DirectoryException
     {
       if (sendEntryData.persistentSearchCanSendEntry(changeNumber))
       {
-        sendEntryIfMatches(searchOp, entry, null);
+        return sendNotificationIfMatches(searchOp, entry, null);
       }
+      return true;
     }
   }
 
@@ -1713,7 +1772,11 @@
       return sendEntryIfMatches(searchOp, entry, cookieString);
     }
 
-    private void persistentSearchSendEntry(DN baseDN, UpdateMsg updateMsg)
+    /**
+     * @return {@code true} if the persistent search should keep reporting changes, {@code false}
+     *         otherwise
+     */
+    private boolean persistentSearchSendEntry(DN baseDN, UpdateMsg updateMsg)
         throws DirectoryException
     {
       final CSN csn = updateMsg.getCSN();
@@ -1725,8 +1788,9 @@
         final Entry cookieEntry = createEntryFromMsg(baseDN, 0, cookieString, updateMsg);
         // FIXME JNR use this instead of previous line:
         // entry.replaceAttribute(Attributes.create("changelogcookie", cookieString));
-        sendEntryIfMatches(searchOp, cookieEntry, cookieString);
+        return sendNotificationIfMatches(searchOp, cookieEntry, cookieString);
       }
+      return true;
     }
 
     private String updateCookie(DN baseDN, final CSN csn)

--
Gitblit v1.10.0