Change deleteStore to check StoreHeader before bumping MetaDataVersionStamp

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/FDBRecordStore.java

@@ -1798,23 +1798,83 @@
      * data this will delete the store's header and therefore will remove any evidence that
      * the store existed.
      *
      * <p>
-     * This method does not read the underlying record store, so it does not validate
-     * that a record store exists in the given subspace. As it might be the case that
-     * this record store has a cacheable store state (see {@link #setStateCacheability(boolean)}),
-     * this method resets the database's
-     * {@linkplain FDBRecordContext#getMetaDataVersionStamp(IsolationLevel) meta-data version-stamp}.
-     * As a result, calling this method may cause other clients to invalidate their caches needlessly.
+     * This method reads only the record store's header key (see {@link #STORE_INFO_KEY}) in
+     * order to decide whether it needs to invalidate cached state. If a header is present and
+     * marks the store as {@linkplain #setStateCacheability(boolean) cacheable}, the database's
+     * {@linkplain FDBRecordContext#getMetaDataVersionStamp(IsolationLevel) meta-data
+     * version-stamp} is reset so that other clients drop their cached copies; if the header is
+     * missing (store doesn't exist) or marks the store as non-cacheable, the version-stamp is
+     * not touched. This means callers who only ever operate on non-cacheable stores do not
+     * contend on the single meta-data version-stamp key.
+     * </p>
+     *
+     * <p>
+     * The header read uses snapshot isolation to avoid adding a read-conflict range on every
+     * delete. Only when the on-disk header says the store is non-cacheable (or is missing) —
+     * i.e., the case in which we would skip the bump — do we add an explicit point read
+     * conflict on {@link #STORE_INFO_KEY}. That closes the race with a concurrent
+     * {@link #setStateCacheability(boolean)} that flips the store to cacheable: if such a
+     * writer commits before us, our commit fails with a serialization error and the caller
+     * retries with a fresh snapshot that sees the new header and bumps the stamp. In the
+     * common case (delete of a store that was and stays non-cacheable), no read conflict is
+     * added at all.
      * </p>
      *
      * @param context the transactional context in which to delete the record store
      * @param subspace the subspace containing the record store
      */
     @SuppressWarnings("PMD.CloseResource")
     public static void deleteStore(FDBRecordContext context, Subspace subspace) {
-        // In theory, we only need to set the meta-data version stamp if the record store's
-        // meta-data is cacheable, but we can't know that from here.
-        context.setMetaDataVersionStamp();
+        // Read the store header at snapshot isolation so we don't add a read conflict on
+        // every delete. The clear below already creates a write conflict on the whole range,
+        // which serializes us against any concurrent writer that lands INSIDE this subspace.
+        // The single race the snapshot read leaves open is a concurrent commit that writes
+        // STORE_INFO_KEY (e.g. a setStateCacheability that flips cacheable=true and bumps
+        // the meta-data version stamp). If that writer commits first, our snapshot read
+        // misses the flip and we would skip our own bump — leaving sibling caches populated
+        // from the writer's committed state stale after our delete.
+        //
+        // We close that race narrowly: only on the branches where we decide NOT to bump
+        // (header absent, or header parses as non-cacheable), we add an explicit point read
+        // conflict on STORE_INFO_KEY. That way:
+        //   - the common case (delete of a store that was and stays non-cacheable) still
+        //     contributes zero read conflicts, and
+        //   - the racy case (concurrent flip to cacheable) is detected: the writer's SET on
+        //     STORE_INFO_KEY overlaps our added read conflict and we lose the commit race.
+        final byte[] headerKey = subspace.pack(STORE_INFO_KEY);
+        final byte[] headerBytes = context.asyncToSync(FDBStoreTimer.Waits.WAIT_LOAD_RECORD_STORE_STATE,
+                context.readTransaction(true).get(headerKey));
+        boolean shouldBump;
+        if (headerBytes == null) {
+            // Header absent: no cached state exists to invalidate — no bump needed. But guard
+            // against a concurrent creator/enabler flipping the store to cacheable underneath
+            // us: adding a read conflict on STORE_INFO_KEY makes such a writer's commit
+            // exclude ours.
+            context.ensureActive().addReadConflictKey(headerKey);
+            shouldBump = false;
+        } else {
+            boolean cacheable;
+            try {
+                cacheable = RecordMetaDataProto.DataStoreInfo.parseFrom(headerBytes).getCacheable();
+            } catch (InvalidProtocolBufferException e) {
+                // If we can't parse the header, fall back to the conservative
+                // behavior and bump. Also skip the read conflict — bumping unconditionally
+                // is safe regardless of what a concurrent writer does.
+                cacheable = true;
+            }
+            if (cacheable) {
+                shouldBump = true;
+            } else {
+                // Header says non-cacheable AND parsed cleanly. If a concurrent writer flips
+                // it to cacheable, we must lose the commit race so the caller retries.
+                context.ensureActive().addReadConflictKey(headerKey);
+                shouldBump = false;
+            }
+        }
+        if (shouldBump) {
+            context.setMetaDataVersionStamp();
+        }
         context.setDirtyStoreState(true);
         context.clear(subspace.range());
     }

fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/storestate/FDBRecordStoreStateCacheTest.java

@@ -1,4 +1,4 @@
 /*
  * FDBRecordStoreStateCacheTest.java
  *
  * This source file is part of the FoundationDB open source project
@@ -25,8 +25,8 @@
 import com.apple.foundationdb.record.RecordCoreException;
 import com.apple.foundationdb.record.RecordMetaData;
 import com.apple.foundationdb.record.RecordMetaDataBuilder;
-import com.apple.foundationdb.record.expressions.RecordKeyExpressionProto;
 import com.apple.foundationdb.record.TestRecords1Proto;
+import com.apple.foundationdb.record.expressions.RecordKeyExpressionProto;
 import com.apple.foundationdb.record.metadata.Key;
 import com.apple.foundationdb.record.metadata.expressions.KeyExpression;
 import com.apple.foundationdb.record.provider.foundationdb.FDBDatabase;
@@ -45,6 +45,7 @@
 import com.apple.foundationdb.record.provider.foundationdb.RecordStoreStaleMetaDataVersionException;
 import com.apple.foundationdb.record.provider.foundationdb.keyspace.KeySpacePath;
 import com.apple.foundationdb.record.test.FakeClusterFileUtil;
+import com.apple.foundationdb.subspace.Subspace;
 import com.apple.foundationdb.tuple.ByteArrayUtil;
 import com.apple.foundationdb.tuple.ByteArrayUtil2;
 import com.apple.foundationdb.tuple.Tuple;
@@ -56,6 +57,7 @@
 import org.junit.jupiter.params.provider.MethodSource;
 
 import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.UUID;
@@ -75,6 +77,7 @@
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
 
 /**
  * Tests to make sure that caching {@link FDBRecordStoreStateCacheEntry} objects work.
@@ -766,6 +769,199 @@
         }
     }
 
+    /**
+     * Deleting a non-cacheable store must NOT bump the meta-data version stamp: no other
+     * client can possibly hold a cached copy of a non-cacheable header, so the version stamp
+     * (a JVM-wide bottleneck on the {@code \xff/metadataVersion} key) shouldn't be touched.
+     * This means that deleting a store won't cause any interaction with any cacheable store to conflict.
+     */
+    @Test
+    void deleteNonCacheableStoreDoesNotBumpMetaDataVersionStamp() throws Exception {
+        ensureMetaDataVersionStampInitialized();
+
+        Subspace subspace;
+        try (FDBRecordContext context = openContext()) {
+            openSimpleRecordStore(context);
+            assertNotCacheable();
+            subspace = recordStore.getSubspace();
+            commit(context);
+        }
+
+        deleteStoreDoesNotBumpMetaDataVersion(subspace);
+    }
+
+    /**
+     * Deleting an empty subspace (no store header present) must not bump the stamp either —
+     * there's no cached header to invalidate.
+     */
+    @Test
+    void deleteMissingStoreDoesNotBumpMetaDataVersionStamp() {
+        ensureMetaDataVersionStampInitialized();
+
+        // Use the test's per-instance path, but never open a store there.
+        final Subspace subspace;
+        try (FDBRecordContext context = openContext()) {
+            subspace = path.toSubspace(context);
+        }
+        deleteStoreDoesNotBumpMetaDataVersion(subspace);
+    }
+
+    private void deleteStoreDoesNotBumpMetaDataVersion(final Subspace subspace) {
+        final byte[] beforeStamp = getMetaDataVersionStamp();
+        assertNotNull(beforeStamp);
+
+        deleteStore(subspace);
+
+        final byte[] afterStamp = getMetaDataVersionStamp();
+        assertNotNull(afterStamp);
+        assertArrayEquals(beforeStamp, afterStamp,
+                "deleting a cacheable store should not have bumped the meta-data version stamp");
+    }
+
+    /**
+     * Complement of {@link #deleteNonCacheableStoreDoesNotBumpMetaDataVersionStamp()}: deleting
+     * a cacheable store MUST bump the stamp — otherwise sibling clients could keep serving
+     * reads out of a stale cached header long after the store is gone.
+     */
+    @Test
+    void deleteCacheableStoreBumpsMetaDataVersionStamp() throws Exception {
+        ensureMetaDataVersionStampInitialized();
+
+        Subspace subspace;
+        try (FDBRecordContext context = openContext()) {
+            openSimpleRecordStore(context);
+            assertTrue(recordStore.setStateCacheability(true), "flipping to cacheable should have changed something");
+            subspace = recordStore.getSubspace();
+            commit(context);
+        }
+        // Commit above already bumped the stamp (transition to cacheable). Snapshot after that.
+        final byte[] beforeStamp = getMetaDataVersionStamp();
+        assertNotNull(beforeStamp);
+
+        deleteStore(subspace);
+
+        final byte[] afterStamp = getMetaDataVersionStamp();
+        assertNotNull(afterStamp);
+        assertFalse(Arrays.equals(beforeStamp, afterStamp),
+                "deleting a cacheable store should have bumped the meta-data version stamp");
+    }
+
+    /**
+     * Race check for {@link FDBRecordStore#deleteStore(FDBRecordContext, Subspace)}.
+     * <p>The bug we're guarding against: {@code deleteStore} commits without bumping the
+     * stamp, concurrently with another transaction that sets the store as cacheable.
+     * The delete should conflict.</p>
+     */
+    @Test
+    void concurrentSetCacheabilityPreventsDeleteStore() throws Exception {
+        fdb.setStoreStateCache(metaDataVersionStampCacheFactory.getCache(fdb));
+        ensureMetaDataVersionStampInitialized();
+
+        // Setup: create the store as NON-cacheable. That is the trap for the deleter's
+        // snapshot header read.
+        FDBRecordStore.Builder storeBuilder;
+        Subspace subspace;
+        try (FDBRecordContext context = openContext()) {
+            openSimpleRecordStore(context);
+            assertNotCacheable();
+            storeBuilder = recordStore.asBuilder();
+            subspace = recordStore.getSubspace();
+            commit(context);
+        }
+
+        final boolean deleterCommitted;
+        try (FDBRecordContext deleterContext = fdb.openContext(null, new FDBStoreTimer())) {
+            // Pin the deleter's read version BEFORE the flip commits, so the deleter's
+            // snapshot header read sees the pre-flip (non-cacheable) state.
+            deleterContext.getReadVersion();
+
+            // Separate transaction: flip the store to cacheable. Kept separate from the
+            // record insert below so that the record-inserting transaction has no reason to
+            // read the header itself — it will pick up the cacheable header via the state
+            // cache.
+            try (FDBRecordContext flipperContext = fdb.openContext(null, new FDBStoreTimer())) {
+                FDBRecordStore flipperStore = storeBuilder.copyBuilder().setContext(flipperContext).open();
+                assertTrue(flipperStore.setStateCacheability(true),
+                        "flipping to cacheable should have changed the header");
+                commit(flipperContext);
+            }
+
+            // deleterContext: reads STORE_INFO_KEY at snapshot at its pinned pre-flip RV, sees
+            // non-cacheable. The deleteStore also adds a point read conflict
+            // on STORE_INFO_KEY — which the flipper's committed write will conflict with.
+            FDBRecordStore.deleteStore(deleterContext, subspace);
+            deleterCommitted = tryCommitOrDetectConflict(deleterContext);
+        }
+
+        if (deleterCommitted) {
+            fail("delete committed after a concurrent setStateCacheability(true) flip " +
+                    "and did not bump the meta-data version stamp; a subsequent writer " +
+                    "could have trusted its (now-stale) cache entry and inserted a record into a " +
+                    "subspace whose store header has been deleted — an on-disk orphan. " +
+                    "This is the correctness bug deleteStore's read-conflict guard is meant " +
+                    "to prevent.");
+        } else {
+            // Deleter correctly conflicted with the flipper. The store still exists (the
+            // flipper's cacheable header); caches remain consistent with disk.
+            try (FDBRecordContext contextUsingCache = fdb.openContext(null, new FDBStoreTimer())) {
+                // ensure that we can open the store, guaranteeing the store header still exists
+                recordStore = storeBuilder.copyBuilder().setContext(contextUsingCache).open();
+                assertCacheable();
+            }
+        }
+    }
+
+    /**
+     * Companion invariant to {@link #concurrentSetCacheabilityPreventsDeleteStore()}: with the
+     * commit order flipped (deleter first, then flipper), the flipper must conflict. Unlike its
+     * companion, this test relies on general read/write conflict machinery — specifically, that
+     * setStateCacheability's own SERIALIZABLE header read (via open() → loadRecordStoreStateAsync
+     * or handleCachedState) puts STORE_INFO_KEY in the flipper's read set, so any concurrent
+     * write to that key by a preceding delete forces a conflict.
+     */
+    @Test
+    void concurrentSetCacheabilityConflictsWithDeleteStore() throws Exception {
+        fdb.setStoreStateCache(metaDataVersionStampCacheFactory.getCache(fdb));
+        ensureMetaDataVersionStampInitialized();
+
+        // Setup: create the store as NON-cacheable (matches the companion test's starting state).
+        FDBRecordStore.Builder storeBuilder;
+        Subspace subspace;
+        try (FDBRecordContext context = openContext()) {
+            openSimpleRecordStore(context);
+            assertNotCacheable();
+            storeBuilder = recordStore.asBuilder();
+            subspace = recordStore.getSubspace();
+            commit(context);
+        }
+
+        final boolean flipperCommitted;
+        try (FDBRecordContext flipperContext = fdb.openContext(null, new FDBStoreTimer())) {
+            FDBRecordStore flipperStore = storeBuilder.copyBuilder().setContext(flipperContext).open();
+            try (FDBRecordContext deleterContext = fdb.openContext(null, new FDBStoreTimer())) {
+                FDBRecordStore.deleteStore(deleterContext, subspace);
+                commit(deleterContext);
+            }
+            assertTrue(flipperStore.setStateCacheability(true),
+                    "flipping to cacheable should have changed the header");
+            flipperCommitted = tryCommitOrDetectConflict(flipperContext);
+        }
+
+        if (flipperCommitted) {
+            fail("a setStateCacheability(true) flip committed after a successful deleteStore; " +
+                    "another transaction could have used the cached state to write after the " +
+                    "delete store without there being a store header");
+        } else {
+            // Deleter correctly conflicted with the flipper. The store still should have been deleted.
+            try (FDBRecordContext contextUsingCache = fdb.openContext(null, new FDBStoreTimer())) {
+                assertTrue(contextUsingCache.ensureActive().getRange(subspace.range()).asList().join().isEmpty(),
+                        "The store should have been deleted");
+                recordStore = storeBuilder.copyBuilder().setContext(contextUsingCache).create();
+                assertNotCacheable();
+            }
+        }
+    }
+
     /**
      * Verify that updating a header user field will be updated if the store state is cached.
      */
@@ -982,12 +1178,7 @@
                 .getCache(fdb);
         fdb.setStoreStateCache(storeStateCache);
 
-        try (FDBRecordContext context = fdb.openContext()) {
-            if (context.getMetaDataVersionStamp(IsolationLevel.SNAPSHOT) == null) {
-                context.setMetaDataVersionStamp();
-            }
-            commit(context);
-        }
+        ensureMetaDataVersionStampInitialized();
 
         // Create the store, initially not cacheable
         FDBStoreTimer timer = new FDBStoreTimer();
@@ -1254,4 +1445,50 @@
     private boolean isStoreCachable() {
         return recordStore.getRecordStoreState().getStoreHeader().getCacheable();
     }
+
+    /**
+     * Bootstrap that guarantees the cluster-wide meta-data version stamp key exists, so callers
+     * can compare before/after snapshots without special-casing null.
+     */
+    private void ensureMetaDataVersionStampInitialized() {
+        try (FDBRecordContext context = fdb.openContext()) {
+            if (context.getMetaDataVersionStamp(IsolationLevel.SNAPSHOT) == null) {
+                context.setMetaDataVersionStamp();
+            }
+            commit(context);
+        }
+    }
+
+    @Nullable
+    private byte[] getMetaDataVersionStamp() {
+        try (FDBRecordContext context = fdb.openContext()) {
+            return context.getMetaDataVersionStamp(IsolationLevel.SNAPSHOT);
+        }
+    }
+
+    private void deleteStore(final Subspace subspace) {
+        try (FDBRecordContext context = openContext()) {
+            FDBRecordStore.deleteStore(context, subspace);
+            commit(context);
+        }
+    }
+
+    /**
+     * Attempt to commit the given context. Returns {@code true} on success, {@code false} if
+     * the commit failed with a conflict (any nested cause). Any other exception is re-raised
+     * so an unexpected failure mode doesn't silently masquerade as "conflict".
+     */
+    private boolean tryCommitOrDetectConflict(@Nonnull FDBRecordContext context) throws Exception {
+        try {
+            commit(context);
+            return true;
+        } catch (Exception ex) {
+            for (Throwable cause = ex; cause != null; cause = cause.getCause()) {
+                if (cause instanceof FDBExceptions.FDBStoreTransactionConflictException) {
+                    return false;
+                }
+            }
+            throw new AssertionError("unexpected exception from commit: " + ex, ex);
+        }
+    }
 }