Enable test concurrency by default

fdb-record-layer-core/fdb-record-layer-core.gradle

@@ -25,6 +25,13 @@ plugins {
 apply from: rootProject.file('gradle/proto.gradle')
 apply from: rootProject.file('gradle/publishing.gradle')
 
+configurations {
+    // Resolved to the mockito-core jar so it can be installed as a -javaagent on the test JVM.
+    // Required on JDK 21+ because Mockito's inline mock maker can no longer self-attach.
+    // See https://javadoc.io/doc/org.mockito/mockito-core/latest/org.mockito/org/mockito/Mockito.html#0.3
+    mockitoAgent
+}
+
 dependencies {
     annotationProcessor project(':fdb-java-annotations')
     api project(':fdb-extensions')
@@ -60,6 +67,12 @@ dependencies {
     testFixturesImplementation(libs.protobuf.util)
     testFixturesCompileOnly(libs.jsr305)
     testFixturesAnnotationProcessor(libs.autoService)
+
+    mockitoAgent(libs.mockito) { transitive = false }
+}
+
+tasks.withType(Test).configureEach {
+    jvmArgs("-javaagent:${configurations.mockitoAgent.asPath}")
 }
 
 sourceSets {

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

@@ -1799,22 +1799,42 @@ public static void deleteStore(FDBRecordContext context, KeySpacePath path) {
      * 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. The header read uses snapshot
+     * isolation and does not add a read conflict, so it does not narrow the write-conflict
+     * behavior of the clear that follows.
      * </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 needless read conflict:
+        // the clear below already creates a write conflict on the whole range, which is what
+        // actually serialises us against any concurrent writer of this store's header.
+        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));
+        if (headerBytes != null) {
+            boolean shouldBump = true;
+            try {
+                shouldBump = RecordMetaDataProto.DataStoreInfo.parseFrom(headerBytes).getCacheable();
+            } catch (InvalidProtocolBufferException e) {
+                // If we can't parse the header, fall back to the pre-change conservative
+                // behavior and bump. This preserves the "delete a store I know nothing about"
+                // contract this method advertises.
+            }
+            if (shouldBump) {
+                context.setMetaDataVersionStamp();
+            }
+        }
         context.setDirtyStoreState(true);
         context.clear(subspace.range());
     }
@@ -3019,9 +3039,34 @@ private CompletableFuture<Void> updateStoreHeaderAsync(@Nonnull UnaryOperator<Re
 
     @Nonnull
     private static CompletableFuture<KeyValue> readStoreFirstKey(@Nonnull FDBRecordContext context, @Nonnull Subspace subspace, @Nonnull IsolationLevel isolationLevel) {
-        final AsyncIterator<KeyValue> iterator = context.readTransaction(isolationLevel.isSnapshot()).getRange(subspace.range(), 1).iterator();
+        // Always read at snapshot isolation so FDB does not add an implicit read-conflict range
+        // spanning the whole subspace (which can pull in unrelated concurrent writers deep inside
+        // the store's subspace and produce spurious SERIALIZATION_FAILUREs). When the caller
+        // wanted SERIALIZABLE isolation, add back a narrower, deterministic read-conflict range
+        // ourselves: from the beginning of the subspace up to and including the first key we
+        // actually found. When the store is empty (no keys at all) we still have to cover the
+        // whole range to correctly serialise against a concurrent creator, so we widen back to
+        // the full subspace in that case.
+        final Range subspaceRange = subspace.range();
+        final AsyncIterator<KeyValue> iterator = context.readTransaction(true).getRange(subspaceRange, 1).iterator();
         return context.instrument(FDBStoreTimer.Events.LOAD_RECORD_STORE_INFO,
-                iterator.onHasNext().thenApply(hasNext -> hasNext ? iterator.next() : null));
+                iterator.onHasNext().thenApply(hasNext -> {
+                    final KeyValue firstKey = hasNext ? iterator.next() : null;
+                    if (!isolationLevel.isSnapshot()) {
+                        final byte[] end;
+                        if (firstKey == null) {
+                            // Store is empty — we need to serialise against anyone else who might
+                            // write into this subspace between our read and commit.
+                            end = subspaceRange.end;
+                        } else {
+                            // Cover [subspaceRange.begin, firstKey + \x00) — i.e. the found key
+                            // itself and everything before it, but nothing after.
+                            end = ByteArrayUtil.join(firstKey.getKey(), new byte[]{0});
+                        }
+                        context.ensureActive().addReadConflictRange(subspaceRange.begin, end);
+                    }
+                    return firstKey;
+                }));
     }
 
     /**

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

@@ -30,6 +30,10 @@
 import com.apple.foundationdb.annotation.API;
 import com.apple.foundationdb.async.AsyncUtil;
 import com.apple.foundationdb.record.provider.common.StoreTimer;
+import com.apple.foundationdb.tuple.ByteArrayUtil;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -39,10 +43,21 @@
 /**
  * Wrapper around {@link Transaction} that instruments certain calls to expose their behavior with
  * {@link FDBStoreTimer} metrics.
+ *
+ * <p>When this class's logger is set to {@code TRACE}, every mutation call
+ * ({@link #clear(byte[])}, {@link #clear(byte[], byte[])}, {@link #clear(Range)},
+ * {@link #clearRangeStartsWith(byte[])}, and range-based
+ * {@link #addWriteConflictRange(byte[], byte[])}) emits the affected key range along with a
+ * captured stack trace. This is intended purely for diagnostic use — for example when hunting
+ * down which code path is issuing a wide subspace clear that's causing a
+ * {@code SERIALIZATION_FAILURE} on a concurrent read. Leaving the logger at its default level
+ * (WARN or above) keeps the overhead a single {@code isTraceEnabled()} check per call.</p>
  */
 @API(API.Status.INTERNAL)
 public class InstrumentedTransaction extends InstrumentedReadTransaction<Transaction> implements Transaction {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(InstrumentedTransaction.class);
+
     @Nullable
     protected ReadTransaction snapshot; // lazily cached snapshot wrapper
 
@@ -77,6 +92,7 @@ public void addReadConflictKey(byte[] key) {
 
     @Override
     public void addWriteConflictRange(byte[] keyBegin, byte[] keyEnd) {
+        traceMutation("addWriteConflictRange", keyBegin, keyEnd);
         underlying.addWriteConflictRange(checkKey(keyBegin), checkKey(keyEnd));
     }
 
@@ -94,19 +110,22 @@ public void set(byte[] key, byte[] value) {
 
     @Override
     public void clear(byte[] key) {
+        traceMutation("clear(key)", key, null);
         underlying.clear(checkKey(key));
         increment(FDBStoreTimer.Counts.DELETES);
     }
 
     @Override
     public void clear(byte[] keyBegin, byte[] keyEnd) {
+        traceMutation("clear(range)", keyBegin, keyEnd);
         underlying.clear(checkKey(keyBegin), checkKey(keyEnd));
         increment(FDBStoreTimer.Counts.DELETES);
         increment(FDBStoreTimer.Counts.RANGE_DELETES);
     }
 
     @Override
     public void clear(Range range) {
+        traceMutation("clear(Range)", range.begin, range.end);
         checkKey(range.begin);
         checkKey(range.end);
 
@@ -118,11 +137,28 @@ public void clear(Range range) {
     @Override
     @Deprecated
     public void clearRangeStartsWith(byte[] prefix) {
+        traceMutation("clearRangeStartsWith", prefix, null);
         underlying.clearRangeStartsWith(checkKey(prefix));
         increment(FDBStoreTimer.Counts.DELETES);
         increment(FDBStoreTimer.Counts.RANGE_DELETES);
     }
 
+    /**
+     * Diagnostic hook: when this class's SLF4J logger is at TRACE, emit the operation name, the
+     * printable form of the affected key(s), and a stack trace. Callers pay only the cost of the
+     * {@code isTraceEnabled()} check at other log levels. Intended for hunting down subspace-wide
+     * range writes that cause serialization failures on concurrent readers.
+     */
+    private static void traceMutation(@Nonnull String op, @Nullable byte[] begin, @Nullable byte[] end) {
+        if (!LOGGER.isTraceEnabled()) {
+            return;
+        }
+        final String beginStr = begin == null ? "null" : ByteArrayUtil.printable(begin);
+        final String endStr = end == null ? "" : ".." + ByteArrayUtil.printable(end);
+        LOGGER.trace("InstrumentedTransaction mutation: {} [{}{}]", op, beginStr, endStr,
+                new Throwable("stack for " + op));
+    }
+
     @Override
     public void mutate(MutationType opType, byte[] key, byte[] param) {
         underlying.mutate(opType, checkKey(key), param);

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

@@ -107,12 +107,19 @@ void cachedVersionMaintenanceOnReadsTest() throws Exception {
 
         RecordMetaData metaData = RecordMetaData.build(TestRecords1Proto.getDescriptor());
 
+        // Create the path up-front so that the directory-layer resolution done by
+        // TestKeySpacePathManager.createPath has already warmed FDBDatabase.lastSeenFDBVersion
+        // by the time we capture readVersion1 below. Otherwise the cached version captured
+        // here would be stale relative to the one createPath puts in the cache, and the
+        // weak-read-semantics assertion at "assertEquals(readVersion1, context.getReadVersion())"
+        // would return the newer (post-createPath) cached version and fail.
+        final KeySpacePath path = pathManager.createPath(TestKeySpace.RECORD_STORE);
+
         // First time, does a GRV from FDB
-        long readVersion1 = getReadVersion(database, 0L, 2_000L);
+        long readVersion1 = getReadVersion(database, 0L, 60_000L);
 
         // Store a record (advances future GRV, but not cached version)
-        final KeySpacePath path = pathManager.createPath(TestKeySpace.RECORD_STORE);
-        final FDBDatabase.WeakReadSemantics weakReadSemantics = new FDBDatabase.WeakReadSemantics(0L, 2_000L, false);
+        final FDBDatabase.WeakReadSemantics weakReadSemantics = new FDBDatabase.WeakReadSemantics(0L, 60_000L, false);
         try (FDBRecordContext context = database.openContext(FDBRecordContextConfig.newBuilder().setWeakReadSemantics(weakReadSemantics).setMdcContext(MDC.getCopyOfContextMap()).build())) {
             assertEquals(readVersion1, context.getReadVersion());
             FDBRecordStore recordStore = FDBRecordStore.newBuilder()
@@ -126,28 +133,41 @@ void cachedVersionMaintenanceOnReadsTest() throws Exception {
             context.commit();
         }
 
-        // We're fine with any version obtained up to 2s ago, so will get the original readVersion
-        assertEquals(readVersion1, getReadVersion(database, 0L, 2000L));
+        // We're fine with any version obtained up to 60s ago, so will get the original readVersion
+        assertEquals(readVersion1, getReadVersion(database, 0L, 60_000L));
 
         Thread.sleep(10L);
 
         // Set a short staleness bound, this will cause a GRV call to FDB and will update the cached version
         long readVersion2 = getReadVersion(database, 0L, 11L);
         assertThat(readVersion1, lessThan(readVersion2));
 
-        // Store another record
-        testStoreAndRetrieveSimpleRecord(database, metaData, path);
+        // Store another record. Disable read-tracking for the duration so the fresh GRV
+        // done inside testStoreAndRetrieveSimpleRecord (which uses database.run(null, ...)
+        // and therefore bypasses weak-read-semantics) doesn't repopulate the cached
+        // version with a newer value than readVersion2 and break the assertions below.
+        database.setTrackLastSeenVersionOnRead(false);
+        try {
+            testStoreAndRetrieveSimpleRecord(database, metaData, path);
+        } finally {
+            database.setTrackLastSeenVersionOnRead(true);
+        }
 
-        assertEquals(readVersion2, getReadVersion(database, 0L, 2000L));
-        assertEquals(readVersion2, getReadVersion(database, readVersion2, 2000L));
+        assertEquals(readVersion2, getReadVersion(database, 0L, 60_000L));
+        assertEquals(readVersion2, getReadVersion(database, readVersion2, 60_000L));
 
         // Now we want at least readVersion2 + 1, so this will cause a GRV
-        long readVersion3 = getReadVersion(database, readVersion2 + 1, 2000L);
+        long readVersion3 = getReadVersion(database, readVersion2 + 1, 60_000L);
 
         assertTrue(readVersion2 < readVersion3);
 
-        // Store another record
-        testStoreAndRetrieveSimpleRecord(database, metaData, path);
+        // Store another record (same read-tracking-disable shield as above)
+        database.setTrackLastSeenVersionOnRead(false);
+        try {
+            testStoreAndRetrieveSimpleRecord(database, metaData, path);
+        } finally {
+            database.setTrackLastSeenVersionOnRead(true);
+        }
 
         // Don't use a stored version
         assertTrue(readVersion3 < getReadVersion(database, null, null));