GH-5807 fix: avoid closing active ValueStore readers

Author: hmottestad

core/sail/lmdb/src/main/java/org/eclipse/rdf4j/sail/lmdb/ValueStore.java

@@ -930,7 +930,7 @@ <T> T readTransaction(long env, Transaction<T> transaction) throws IOException {
 				if (attempt > 0) {
 					throw e;
 				}
-				closeReadTransactions();
+				closeInactiveReadTransactions();
 			}
 		}
 	}
@@ -1404,6 +1404,18 @@ private void closeReadTransactions() {
 		}
 	}
 
+	private void closeInactiveReadTransactions() {
+		txnLock.writeLock().lock();
+		try {
+			ReadTxn.State[] snapshot = activeReadTransactions.toArray(new ReadTxn.State[0]);
+			for (ReadTxn.State readTxn : snapshot) {
+				readTxn.closeInactiveTxn();
+			}
+		} finally {
+			txnLock.writeLock().unlock();
+		}
+	}
+
 	/**
 	 * Closes the ValueStore, releasing any file references, etc. Once closed, the ValueStore can no longer be used.
 	 *
@@ -1540,6 +1552,12 @@ synchronized void closeTxn() {
 					activeReadTransactions.remove(this);
 				}
 			}
+
+			synchronized void closeInactiveTxn() {
+				if (initialized && depth == 0) {
+					closeTxn();
+				}
+			}
 		}
 
 		public ReadTxn(Set<State> activeReadTransactions, ConcurrentCleaner cleaner) {

core/sail/lmdb/src/test/java/org/eclipse/rdf4j/sail/lmdb/ValueStoreTest.java

@@ -28,6 +28,10 @@
 import java.util.List;
 import java.util.Random;
 import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReadWriteLock;
 import java.util.stream.Collectors;
 
 import org.eclipse.rdf4j.model.IRI;
@@ -407,6 +411,58 @@ public void testStaleBNodeHashCodeIgnoresReusedIdAfterClear() throws Exception {
 		assertEquals("stale values must keep their original hash after a revision change", firstHash, first.hashCode());
 	}
 
+	@Test
+	public void testReadersFullCleanupDoesNotCloseInUseReadTxn() throws Exception {
+		ReadWriteLock txnLock = getField(valueStore, "txnLock", ReadWriteLock.class);
+		Set<?> activeReadTransactions = getField(valueStore, "activeReadTransactions", Set.class);
+		long env = getLongField(valueStore, "env");
+
+		CountDownLatch readLockReleased = new CountDownLatch(1);
+		CountDownLatch cleanupAttempted = new CountDownLatch(1);
+		AtomicReference<Throwable> workerFailure = new AtomicReference<>();
+
+		Thread worker = new Thread(() -> {
+			try {
+				valueStore.readTransaction(env, (stack, txn) -> {
+					txnLock.readLock().unlock();
+					try {
+						readLockReleased.countDown();
+						try {
+							assertTrue("cleanup should run", cleanupAttempted.await(10, TimeUnit.SECONDS));
+						} catch (InterruptedException e) {
+							Thread.currentThread().interrupt();
+							throw new IOException(e);
+						}
+					} finally {
+						txnLock.readLock().lock();
+					}
+					return null;
+				});
+			} catch (Throwable t) {
+				workerFailure.set(t);
+			}
+		}, "value-store-read-transaction-gap");
+		worker.start();
+
+		assertTrue("read transaction should reach the resize-style read lock gap",
+				readLockReleased.await(10, TimeUnit.SECONDS));
+		assertEquals("test setup should have one active read transaction", 1, activeReadTransactions.size());
+
+		try {
+			invokeCloseInactiveReadTransactions(valueStore);
+			assertEquals("reader cleanup must not close a read transaction that is still executing", 1,
+					activeReadTransactions.size());
+		} finally {
+			cleanupAttempted.countDown();
+			worker.join(TimeUnit.SECONDS.toMillis(10));
+		}
+
+		assertFalse("worker should finish", worker.isAlive());
+		if (workerFailure.get() != null) {
+			throw new AssertionError(workerFailure.get());
+		}
+	}
+
 	private long storeValueAndReopen(Value value) throws Exception {
 		return storeValueAndReopen(value, new LmdbStoreConfig());
 	}
@@ -438,6 +494,24 @@ private boolean isInitialized(Object value) throws Exception {
 		return initializedField.getBoolean(value);
 	}
 
+	private <T> T getField(Object target, String fieldName, Class<T> fieldType) throws Exception {
+		Field field = target.getClass().getDeclaredField(fieldName);
+		field.setAccessible(true);
+		return fieldType.cast(field.get(target));
+	}
+
+	private long getLongField(Object target, String fieldName) throws Exception {
+		Field field = target.getClass().getDeclaredField(fieldName);
+		field.setAccessible(true);
+		return field.getLong(target);
+	}
+
+	private void invokeCloseInactiveReadTransactions(ValueStore store) throws Exception {
+		var method = store.getClass().getDeclaredMethod("closeInactiveReadTransactions");
+		method.setAccessible(true);
+		method.invoke(store);
+	}
+
 	@AfterEach
 	public void after() throws Exception {
 		try {