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,8 @@
 import java.util.concurrent.atomic.AtomicLong;
 
 import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.nullValue;
 
 public class LiveVersionMapTests extends ESTestCase {
 
@@ -396,6 +398,40 @@ public void testPruneTombstonesWhileLocked() throws InterruptedException, IOExce
         assertEquals(0, map.getAllTombstones().size());
     }
 
+    public void testRandomlyIndexDeleteAndRefresh() throws Exception {
+        final LiveVersionMap versionMap = new LiveVersionMap();
+        final BytesRef uid = uid("1");
+        final long versions = between(10, 1000);
+        VersionValue latestVersion = null;
+        for (long i = 0; i < versions; i++) {
+            if (randomBoolean()) {
+                versionMap.beforeRefresh();
+                versionMap.afterRefresh(randomBoolean());
+            }
+            if (randomBoolean()) {
+                versionMap.enforceSafeAccess();
+            }
+            try (Releasable ignore = versionMap.acquireLock(uid)) {
+                if (randomBoolean()) {
+                    latestVersion = new DeleteVersionValue(randomNonNegativeLong(), randomLong(), randomLong(), randomLong());
+                    versionMap.putDeleteUnderLock(uid, (DeleteVersionValue) latestVersion);
+                    assertThat(versionMap.getUnderLock(uid), equalTo(latestVersion));
+                } else if (randomBoolean()) {
+                    latestVersion = new IndexVersionValue(randomTranslogLocation(), randomNonNegativeLong(), randomLong(), randomLong());
+                    versionMap.maybePutIndexUnderLock(uid, (IndexVersionValue) latestVersion);
+                    if (versionMap.isSafeAccessRequired()) {
+                        assertThat(versionMap.getUnderLock(uid), equalTo(latestVersion));
+                    } else {
+                        assertThat(versionMap.getUnderLock(uid), nullValue());
+                    }
+                }
+                if (versionMap.getUnderLock(uid) != null) {
+                    assertThat(versionMap.getUnderLock(uid), equalTo(latestVersion));
+                }
+            }
+        }
+    }
+
     IndexVersionValue randomIndexVersionValue() {
         return new IndexVersionValue(randomTranslogLocation(), randomNonNegativeLong(), randomNonNegativeLong(), randomNonNegativeLong());
     }