Don't add tombstone in dummy persistence when a newer entry already exists

This better mirrors how Proton actually works, since it's not a multi
version store. Since only the highest timestamped entry for a document
is the one that is ever considered on a node, there's no point in
storing an explicit tombstone that cannot be referenced.

2 files changed, 52 insertions, 5 deletions

diff --git a/persistence/src/vespa/persistence/conformancetest/conformancetest.cpp b/persistence/src/vespa/persistence/conformancetest/conformancetest.cpp
index e03403f601d..fac6b0562a3 100644
--- a/persistence/src/vespa/persistence/conformancetest/conformancetest.cpp
+++ b/persistence/src/vespa/persistence/conformancetest/conformancetest.cpp
@@ -783,6 +783,47 @@ TEST_F(ConformanceTest, testRemoveMulti)
     EXPECT_EQ(15u, removeResult->num_removed());
 }
 
+TEST_F(ConformanceTest, multi_remove_does_not_remove_docs_if_specified_timestamp_is_older_than_stored_doc)
+{
+    document::TestDocMan testDocMan;
+    _factory->clear();
+    PersistenceProviderUP spi(getSpi(*_factory, testDocMan));
+
+    BucketId bucketId1(8, 0x01);
+    Bucket bucket1(makeSpiBucket(bucketId1));
+    spi->createBucket(bucket1);
+
+    std::vector<Document::SP> docs;
+    for (size_t i(0); i < 30; i++) {
+        docs.push_back(testDocMan.createRandomDocumentAtLocation(0x01, i));
+    }
+
+    std::vector<spi::IdAndTimestamp> ids;
+    for (size_t i(0); i < docs.size(); i++) {
+        spi->put(bucket1, Timestamp(i + 200), docs[i]);
+        if (i & 0x1) {
+            ids.emplace_back(docs[i]->getId(), Timestamp(i + 100)); // Note: lower timestamps
+        }
+    }
+
+    auto onDone = std::make_unique<CatchResult>();
+    auto future = onDone->future_result();
+    spi->removeAsync(bucket1, ids, std::move(onDone));
+    auto result = future.get();
+    ASSERT_TRUE(result);
+    auto removeResult = dynamic_cast<spi::RemoveResult *>(result.get());
+    ASSERT_TRUE(removeResult != nullptr);
+    EXPECT_EQ(0u, removeResult->num_removed());
+
+    // Nothing shall have been removed
+    Context context(Priority(0), Trace::TraceLevel(0));
+    for (size_t i(0); i < docs.size(); i++) {
+        GetResult getResult = spi->get(bucket1, document::AllFields(), docs[i]->getId(), context);
+        EXPECT_EQ(Result::ErrorType::NONE, getResult.getErrorCode());
+        EXPECT_EQ(Timestamp(i + 200), getResult.getTimestamp());
+    }
+}
+
 TEST_F(ConformanceTest, testRemoveMerge)
 {
     document::TestDocMan testDocMan;
diff --git a/persistence/src/vespa/persistence/dummyimpl/dummypersistence.cpp b/persistence/src/vespa/persistence/dummyimpl/dummypersistence.cpp
index 6eabadc2f86..ccdbf01a81f 100644
--- a/persistence/src/vespa/persistence/dummyimpl/dummypersistence.cpp
+++ b/persistence/src/vespa/persistence/dummyimpl/dummypersistence.cpp
@@ -493,13 +493,19 @@ DummyPersistence::removeAsync(const Bucket& b, std::vector<spi::IdAndTimestamp>
             bc = acquireBucketWithLock(b);
         }
         DocEntry::SP entry((*bc)->getEntry(id));
-        numRemoves += (entry.get() && !entry->isRemove()) ? 1 : 0;
-        auto remEntry = DocEntry::create(t, DocumentMetaEnum::REMOVE_ENTRY, id);
+        if (!entry || entry->getTimestamp() <= t) {
+            numRemoves += (entry && !entry->isRemove()) ? 1 : 0;
+            auto remEntry = DocEntry::create(t, DocumentMetaEnum::REMOVE_ENTRY, id);
 
-        if ((*bc)->hasTimestamp(t)) {
-            (*bc)->eraseEntry(t);
+            if ((*bc)->hasTimestamp(t)) {
+                (*bc)->eraseEntry(t);
+            }
+            (*bc)->insert(std::move(remEntry));
+        } else {
+            LOG(debug, "Not adding tombstone for %s at %" PRIu64 " since it has already "
+                       "been succeeded by a newer write at timestamp %" PRIu64,
+                id.toString().c_str(), t.getValue(), entry->getTimestamp().getValue());
         }
-        (*bc)->insert(std::move(remEntry));
     }
     bc.reset();
     onComplete->onComplete(std::make_unique<RemoveResult>(numRemoves));