Handle delayed calls to notifyRemoveDone() in gid to lid change handler.

3 files changed, 73 insertions, 8 deletions

diff --git a/searchcore/src/tests/proton/reference/gid_to_lid_change_handler/gid_to_lid_change_handler_test.cpp b/searchcore/src/tests/proton/reference/gid_to_lid_change_handler/gid_to_lid_change_handler_test.cpp
index 6aafb3217c7..263159cefea 100644
--- a/searchcore/src/tests/proton/reference/gid_to_lid_change_handler/gid_to_lid_change_handler_test.cpp
+++ b/searchcore/src/tests/proton/reference/gid_to_lid_change_handler/gid_to_lid_change_handler_test.cpp
@@ -260,6 +260,38 @@ TEST_F("Test that pending removes are merged", Fixture)
     f.removeListeners("testdoc", {});
 }
 
+TEST_F("Test that out of order notifyRemoveDone is handled", Fixture)
+{
+    auto &stats = f.addStats();
+    auto listener = std::make_unique<MyListener>(stats, "test", "testdoc");
+    f.addListener(std::move(listener));
+    f.notifyRemove(toGid(doc1), 20);
+    TEST_DO(stats.assertChanges(0, 1));
+    f.notifyRemove(toGid(doc1), 40);
+    TEST_DO(stats.assertChanges(0, 1));
+    f.notifyRemoveDone(toGid(doc1), 40);
+    TEST_DO(stats.assertChanges(0, 1));
+    f.notifyRemoveDone(toGid(doc1), 20);
+    TEST_DO(stats.assertChanges(0, 1));
+    f.notifyPutDone(toGid(doc1), 12, 50);
+    TEST_DO(stats.assertChanges(1, 1));
+    f.removeListeners("testdoc", {});
+}
+
+TEST_F("Test that out of order notifyPutDone is handled", Fixture)
+{
+    auto &stats = f.addStats();
+    auto listener = std::make_unique<MyListener>(stats, "test", "testdoc");
+    f.addListener(std::move(listener));
+    f.notifyRemove(toGid(doc1), 20);
+    TEST_DO(stats.assertChanges(0, 1));
+    f.notifyPutDone(toGid(doc1), 12, 50);
+    TEST_DO(stats.assertChanges(1, 1));
+    f.notifyRemoveDone(toGid(doc1), 20);
+    TEST_DO(stats.assertChanges(1, 1));
+    f.removeListeners("testdoc", {});
+}
+
 }
 
 TEST_MAIN()
diff --git a/searchcore/src/vespa/searchcore/proton/reference/gid_to_lid_change_handler.cpp b/searchcore/src/vespa/searchcore/proton/reference/gid_to_lid_change_handler.cpp
index 79c4ab0480e..341c5434623 100644
--- a/searchcore/src/vespa/searchcore/proton/reference/gid_to_lid_change_handler.cpp
+++ b/searchcore/src/vespa/searchcore/proton/reference/gid_to_lid_change_handler.cpp
@@ -52,8 +52,13 @@ GidToLidChangeHandler::notifyPutDone(GlobalId gid, uint32_t lid, SerialNum seria
     lock_guard guard(_lock);
     auto itr = _pendingRemove.find(gid);
     if (itr != _pendingRemove.end()) {
-        assert(itr->second > serialNum);
-        return; // Document has already been removed later on
+        auto &entry = itr->second;
+        assert(entry.removeSerialNum != serialNum);
+        if (entry.removeSerialNum > serialNum) {
+            return; // Document has already been removed later on
+        }
+        assert(entry.putSerialNum < serialNum);
+        entry.putSerialNum = serialNum;
     }
     notifyPutDone(gid, lid);
 }
@@ -62,10 +67,16 @@ void
 GidToLidChangeHandler::notifyRemove(GlobalId gid, SerialNum serialNum)
 {
     lock_guard guard(_lock);
-    auto insRes = _pendingRemove.insert(std::make_pair(gid, serialNum));
+    auto insRes = _pendingRemove.insert(std::make_pair(gid, PendingRemoveEntry(serialNum)));
     if (!insRes.second) {
-        assert(insRes.first->second < serialNum);
-        insRes.first->second = serialNum;
+        auto &entry = insRes.first->second;
+        assert(entry.removeSerialNum < serialNum);
+        assert(entry.putSerialNum < serialNum);
+        if (entry.removeSerialNum < entry.putSerialNum) {
+            notifyRemove(gid);
+        }
+        entry.removeSerialNum = serialNum;
+        ++entry.refCount;
     } else {
         notifyRemove(gid);
     }
@@ -76,9 +87,13 @@ GidToLidChangeHandler::notifyRemoveDone(GlobalId gid, SerialNum serialNum)
 {
     lock_guard guard(_lock);
     auto itr = _pendingRemove.find(gid);
-    assert(itr != _pendingRemove.end() && itr->second >= serialNum);
-    if (itr->second == serialNum) {
+    assert(itr != _pendingRemove.end());
+    auto &entry = itr->second;
+    assert(entry.removeSerialNum >= serialNum);
+    if (entry.refCount == 1) {
         _pendingRemove.erase(itr);
+    } else {
+        --entry.refCount;
     }
 }
 
diff --git a/searchcore/src/vespa/searchcore/proton/reference/gid_to_lid_change_handler.h b/searchcore/src/vespa/searchcore/proton/reference/gid_to_lid_change_handler.h
index ea625fba3b3..264ece76eaa 100644
--- a/searchcore/src/vespa/searchcore/proton/reference/gid_to_lid_change_handler.h
+++ b/searchcore/src/vespa/searchcore/proton/reference/gid_to_lid_change_handler.h
@@ -22,10 +22,28 @@ class GidToLidChangeHandler : public std::enable_shared_from_this<GidToLidChange
 {
     using lock_guard = std::lock_guard<std::mutex>;
     using Listeners = std::vector<std::unique_ptr<IGidToLidChangeListener>>;
+    struct PendingRemoveEntry {
+        SerialNum removeSerialNum;
+        SerialNum putSerialNum;
+        uint32_t  refCount;
+
+        PendingRemoveEntry(SerialNum removeSerialNum_)
+            : removeSerialNum(removeSerialNum_),
+              putSerialNum(0),
+              refCount(1)
+        {
+        }
+
+        PendingRemoveEntry()
+            : PendingRemoveEntry(0)
+        {
+        }
+    };
+
     std::mutex _lock;
     Listeners _listeners;
     bool _closed;
-    vespalib::hash_map<GlobalId, SerialNum, GlobalId::hash> _pendingRemove;
+    vespalib::hash_map<GlobalId, PendingRemoveEntry, GlobalId::hash> _pendingRemove;
 
     void notifyPutDone(GlobalId gid, uint32_t lid);
     void notifyRemove(GlobalId gid);