Never leave stale delete tombstones in version map

server/src/main/java/org/elasticsearch/index/engine/LiveVersionMap.java

@@ -312,6 +312,10 @@ void maybePutIndexUnderLock(BytesRef uid, IndexVersionValue version) {
         if (maps.isSafeAccessMode()) {
             putIndexUnderLock(uid, version);
         } else {
+            // Even though we don't store a record of the indexing operation (and mark as unsafe),
+            // we should still remove any previous delete for this uuid (avoid accidental accesses).
+            // Not this should not hurt performance because the tombstone is small (or empty) when unsafe is relevant.
+            removeTombstoneUnderLock(uid);
             maps.current.markAsUnsafe();
             assert putAssertionMap(uid, version);
         }

server/src/test/java/org/elasticsearch/index/engine/InternalEngineTests.java

@@ -1576,6 +1576,23 @@ public void testInternalVersioningOnPrimary() throws IOException {
         assertOpsOnPrimary(ops, Versions.NOT_FOUND, true, engine);
     }
 
+    public void testVersionOnPrimaryWithConcurrentRefresh() throws Exception {
+        List<Engine.Operation> ops = generateSingleDocHistory(false, VersionType.INTERNAL, false, 2, 10, 100);
+        CountDownLatch latch = new CountDownLatch(1);
+        AtomicBoolean running = new AtomicBoolean(true);
+        Thread refreshThread = new Thread(() -> {
+            latch.countDown();
+            while (running.get()) {
+                engine.refresh("test");
+            }
+        });
+        refreshThread.start();
+        latch.await();
+        assertOpsOnPrimary(ops, Versions.NOT_FOUND, true, engine);
+        running.set(false);
+        refreshThread.join();
+    }
+
     private int assertOpsOnPrimary(List<Engine.Operation> ops, long currentOpVersion, boolean docDeleted, InternalEngine engine)
         throws IOException {
         String lastFieldValue = null;

server/src/test/java/org/elasticsearch/index/engine/LiveVersionMapTests.java

@@ -40,6 +40,7 @@
 import java.util.concurrent.atomic.AtomicLong;
 
 import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.equalTo;
 
 public class LiveVersionMapTests extends ESTestCase {
 
@@ -396,6 +397,32 @@ public void testPruneTombstonesWhileLocked() throws InterruptedException, IOExce
         assertEquals(0, map.getAllTombstones().size());
     }
 
+    public void testNeverLeaveStaleDeleteTombstone() throws Exception {
+        LiveVersionMap versionMap = new LiveVersionMap();
+        BytesRef uid = uid("1");
+        long versions = between(10, 1000);
+        for (long version = 1; version <= versions; version++) {
+            if (randomBoolean()) {
+                versionMap.beforeRefresh();
+                versionMap.afterRefresh(randomBoolean());
+            }
+            if (randomBoolean()) {
+                versionMap.enforceSafeAccess();
+            }
+            try (Releasable ignore = versionMap.acquireLock(uid)) {
+                if (randomBoolean()) {
+                    versionMap.putDeleteUnderLock(uid, new DeleteVersionValue(version, 1, 1, 1));
+                } else {
+                    versionMap.maybePutIndexUnderLock(uid, new IndexVersionValue(randomTranslogLocation(), version, 1, 1));
+                }
+                VersionValue storedValue = versionMap.getUnderLock(uid);
+                if (storedValue != null) {
+                    assertThat("Keeping a stale version value", storedValue.version, equalTo(version));
+                }
+            }
+        }
+    }
+
     IndexVersionValue randomIndexVersionValue() {
         return new IndexVersionValue(randomTranslogLocation(), randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong());
     }