The stamp is what lets a save retire entries in caches it has no handle on. Link content is + * cached per session, and {@code GorSession.currentSession} is an InheritableThreadLocal that nothing + * clears, so a read and a save of the same link file can legitimately land on different sessions: the + * save can only invalidate the cache of the session bound to its own thread. Rather than reach every + * cache, a save records the time of the write and any entry stamped before it is dropped on lookup. + * + *
Stamped at the start of the read on purpose. A read that opened its stream before the save
+ * replaced the file holds the pre-write content even though it finished afterwards, so it is the start
+ * of the read, not the end, that says whether the content can be trusted.
+ *
+ * @param content the link file content
+ * @param readStartedNanos {@link System#nanoTime()} taken before the read that produced the content
+ */
+public record CachedLinkContent(String content, long readStartedNanos) {
+}
diff --git a/model/src/main/java/org/gorpipe/gor/driver/linkfile/LinkFile.java b/model/src/main/java/org/gorpipe/gor/driver/linkfile/LinkFile.java
index e103be2b..544d19d5 100644
--- a/model/src/main/java/org/gorpipe/gor/driver/linkfile/LinkFile.java
+++ b/model/src/main/java/org/gorpipe/gor/driver/linkfile/LinkFile.java
@@ -68,7 +68,14 @@ public abstract class LinkFile {
public static final String LINK_FILE_VALIDATE_LOAD = "gor.driver.link.validate.load";
public static final String LINK_FILE_VALIDATE_SAVE = "gor.driver.link.validate.save";
- private static final Cache Dropping the entries we can reach is not enough on its own: the session caches belong to
+ * whichever session was bound to the reading thread, and a resolution already in flight can put its
+ * pre-write content into one of them after this call. The mark covers both -- it outlives the
+ * entries it retires, and it is checked on every lookup, so no cache has to be reachable from here.
+ */
+ private static void invalidateCachedContent(StreamSource source) {
+ var path = source.getFullPath();
+ linkWriteMarks.put(path, System.nanoTime());
+ staticLinkCache.invalidate(path);
+ var session = GorSession.currentSession.get();
+ if (session != null) {
+ session.getCache().getLinkCache().invalidate(path);
+ }
+ }
+
+ private static Cache A simple link file is a bare data path, rewritten so rarely that serving one up to the cache
+ * expiry stale is an accepted trade -- and they are the common case, so this is where the saving is.
+ *
+ * A versioned link file is rewritten in normal operation, by {@code appendEntry} and by
+ * versioned-link GC. A stale entry for one could resolve to a generation that GC has already
+ * deleted, and it would do so silently. Keying on the object length and last modified instead does
+ * not help: that metadata is itself served from a cache with its own expiry, so the key would be
+ * built from stale values and would match. See ENGKNOW-3770.
+ */
+ private static boolean isSimpleLinkFile(String content) {
+ return !Strings.isNullOrEmpty(content)
+ && LinkFileV0.VERSION.equals(LinkFileMeta.createOrLoad(content, null, false).getVersion());
}
private static String readLimitedLinkContent(StreamSource source) {
@@ -464,6 +526,18 @@ private static String readLimitedLinkContent(StreamSource source) {
return content;
} catch (IOException e) {
throw new GorResourceException("Failed to read link file: " + source.getFullPath(), source.getFullPath(), e);
+ } finally {
+ // Release the handle this method acquired. Sources reopen lazily (FileSource.open goes
+ // through ensureOpenForRead, S3Source.close is a no-op), so this is transparent to anything
+ // that uses the source afterwards -- LinkFile.save reopens it for writing. While the cache
+ // was keyed on the source object it pinned every source it was handed, which hid the fact
+ // that callers do not close them; keyed on the path, they become collectable and an unclosed
+ // read handle would survive only until the garbage collector ran the finalizer.
+ try {
+ source.close();
+ } catch (Exception e) {
+ log.debug("Failed to close link file source {} after reading", source.getFullPath(), e);
+ }
}
}
diff --git a/model/src/main/java/org/gorpipe/gor/driver/linkfile/LinkFileCacheStats.java b/model/src/main/java/org/gorpipe/gor/driver/linkfile/LinkFileCacheStats.java
new file mode 100644
index 00000000..79c6f6b8
--- /dev/null
+++ b/model/src/main/java/org/gorpipe/gor/driver/linkfile/LinkFileCacheStats.java
@@ -0,0 +1,174 @@
+package org.gorpipe.gor.driver.linkfile;
+
+import org.gorpipe.base.config.PropsHelper;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
+
+/**
+ * Observe-only tally of link file resolutions (ENGKNOW-3770).
+ *
+ * Off unless {@code gor.driver.link.cache.stats} is set, and it never influences what a caller is
+ * served. It answers two questions that decide the shape of the link content cache:
+ *
+ * The counters are approximate under concurrency: two threads resolving the same path at the same
+ * moment can each see the other's content as a change. That is acceptable for a measurement tally and
+ * is not worth locking a read path over.
+ */
+public final class LinkFileCacheStats {
+
+ static final String STATS_ENABLED_KEY = "gor.driver.link.cache.stats";
+ static final String MAX_PATHS_KEY = "gor.driver.link.cache.stats.maxpaths";
+ private static final int DEFAULT_MAX_PATHS = 50000;
+
+ private static final org.slf4j.Logger log = org.slf4j.LoggerFactory.getLogger(LinkFileCacheStats.class);
+
+ // Per version totals. Always counted, even for paths the cap stopped us tracking individually.
+ private static final Map The cache is keyed on the link file path and expires after 5 minutes. That is safe for **simple**
+ * link files -- a bare data path, rewritten so rarely that serving one up to 5 minutes stale is an
+ * accepted trade -- and it is what makes the cache useful at all, since simple links are by far the
+ * common case.
+ *
+ * It is *not* safe for **versioned** link files, which are rewritten in normal operation: a stale
+ * entry could resolve to a generation that versioned-link GC has already deleted, silently. Those are
+ * always re-read. See the ticket for why keying on cached metadata instead does not fix this.
+ *
+ * Before this change the cache was keyed on the {@link org.gorpipe.gor.driver.providers.stream.sources.StreamSource}
+ * object identity and could never hit at all, so nothing was cached in practice.
+ */
+public class UTestLinkFileCache {
+
+ private static final String SIMPLE = "source/data.gorz\n";
+ private static final String SIMPLE_REWRITTEN = "source/rewritten.gorz\n";
+
+ private static final String VERSIONED = """
+ ## SERIAL = 1
+ ## VERSION = 1
+ #FILE\tTIMESTAMP\tMD5\tSERIAL\tINFO
+ source/versions/generation_1.gorz\t2026-01-01T00:00:00.000Z\tMD5SUM1\t1\t
+ """;
+
+ private static final String VERSIONED_REWRITTEN = """
+ ## SERIAL = 2
+ ## VERSION = 1
+ #FILE\tTIMESTAMP\tMD5\tSERIAL\tINFO
+ source/versions/generation_2.gorz\t2026-02-01T00:00:00.000Z\tMD5SUM2\t2\t
+ """;
+
+ @Rule
+ public TemporaryFolder workDir = new TemporaryFolder();
+
+ private Path linkPath;
+
+ @Before
+ public void setUp() {
+ linkPath = workDir.getRoot().toPath().toAbsolutePath().resolve("test.gorz.link");
+ }
+
+ @After
+ public void tearDown() {
+ // Set by the session tests below. It is an InheritableThreadLocal that nothing clears, so
+ // leaving it set would hand this thread's session to every later test in the same JVM.
+ GorSession.currentSession.remove();
+ }
+
+ private String read() throws Exception {
+ return LinkFile.loadContentFromSource(new FileSource(linkPath.toString()));
+ }
+
+ private void saveLinkFile(String link) {
+ try {
+ var linkFile = LinkFile.load(new FileSource(linkPath.toString()));
+ linkFile.appendEntry(link, "NEWMD5SUM");
+ linkFile.save(new DriverBackedFileReader(
+ null, workDir.getRoot().toPath().toAbsolutePath().toString()));
+ } catch (IOException e) {
+ throw new UncheckedIOException(e);
+ }
+ }
+
+ /** A session with its own link cache, bound to the calling thread the way a real one is. */
+ private static GorSession newSessionOnThisThread(String requestId) {
+ var cache = new GorSessionCache();
+ var session = new GorSession(requestId) {
+ @Override
+ public GorSessionCache getCache() {
+ return cache;
+ }
+ };
+ GorSession.currentSession.set(session);
+ return session;
+ }
+
+ @Test
+ public void simpleLinkFileIsServedFromCache() throws Exception {
+ Files.writeString(linkPath, SIMPLE);
+ var firstRead = read();
+ assertEquals(SIMPLE, firstRead);
+
+ Files.writeString(linkPath, SIMPLE_REWRITTEN);
+
+ assertEquals("a simple link file must be served from cache within the expiry window",
+ firstRead, read());
+ }
+
+ @Test
+ public void versionedLinkFileIsNeverCached() throws Exception {
+ Files.writeString(linkPath, VERSIONED);
+ assertEquals(VERSIONED, read());
+
+ Files.writeString(linkPath, VERSIONED_REWRITTEN);
+
+ assertEquals("a versioned link file must be re-read, never served from cache",
+ VERSIONED_REWRITTEN, read());
+ }
+
+ /**
+ * The cache tolerates a link file rewritten by another process going unnoticed until it expires.
+ * It must never do that for one rewritten here: the caller would read back content it just replaced.
+ */
+ @Test
+ public void savingALinkFileInvalidatesItsCachedContent() throws Exception {
+ Files.writeString(linkPath, SIMPLE);
+ assertEquals(SIMPLE, read());
+
+ var linkFile = LinkFile.load(new FileSource(linkPath.toString()));
+ linkFile.appendEntry("source/appended.gorz", "NEWMD5SUM");
+ linkFile.save(new DriverBackedFileReader(null, workDir.getRoot().toPath().toAbsolutePath().toString()));
+
+ assertEquals("content written in this process must not be served from cache",
+ Files.readString(linkPath), read());
+ }
+
+
+ /**
+ * A resolution that is already in flight when a save lands must not leave its content behind. The
+ * stream it read from was opened before the save replaced the file, so what it holds is the
+ * pre-write content -- and it caches that only after the save has dropped everything it could reach.
+ */
+ @Test
+ public void contentReadWhileASaveWasRunningIsNotLeftInTheCache() throws Exception {
+ Files.writeString(linkPath, SIMPLE);
+
+ var saved = new AtomicBoolean();
+ var source = new FileSource(linkPath.toString()) {
+ @Override
+ public InputStream open() {
+ var is = super.open();
+ if (saved.compareAndSet(false, true)) {
+ saveLinkFile("source/appended.gorz");
+ }
+ return is;
+ }
+ };
+
+ assertEquals("the in-flight read holds the content the file had when it opened the stream",
+ SIMPLE, LinkFile.loadContentFromSource(source));
+
+ assertEquals("content read while a save was in flight must not outlive the save",
+ Files.readString(linkPath), read());
+ }
+
+ /**
+ * {@link GorSession#currentSession} is an InheritableThreadLocal that nothing clears, so a read and
+ * a save of the same link file can land on different session caches: the save invalidates the cache
+ * of the session bound to its own thread, which is not the one holding the entry.
+ */
+ @Test
+ public void savingOnAnotherSessionsThreadInvalidatesCachedContent() throws Exception {
+ Files.writeString(linkPath, SIMPLE);
+
+ var readingSession = newSessionOnThisThread("reader");
+ assertEquals(SIMPLE, read());
+
+ var writer = new Thread(() -> {
+ newSessionOnThisThread("writer");
+ saveLinkFile("source/appended.gorz");
+ });
+ writer.start();
+ writer.join();
+
+ GorSession.currentSession.set(readingSession);
+ assertEquals("a save in this process must not leave stale content in another session's cache",
+ Files.readString(linkPath), read());
+ }
+
+ /** An empty link file carries no link and is typically a placeholder about to be written. */
+ @Test
+ public void emptyLinkFileIsNotCached() throws Exception {
+ Files.writeString(linkPath, "");
+ assertEquals("", read());
+
+ Files.writeString(linkPath, SIMPLE);
+
+ assertEquals("an empty link file must not be cached", SIMPLE, read());
+ }
+
+ @Test
+ public void cachedSimpleContentIsKeyedOnThePathNotTheSourceObject() throws Exception {
+ var otherPath = workDir.getRoot().toPath().toAbsolutePath().resolve("other.gorz.link");
+ Files.writeString(linkPath, SIMPLE);
+ Files.writeString(otherPath, SIMPLE_REWRITTEN);
+
+ assertEquals(SIMPLE, read());
+ assertEquals("a different link file must not be served another one's content",
+ SIMPLE_REWRITTEN, LinkFile.loadContentFromSource(new FileSource(otherPath.toString())));
+ }
+}
diff --git a/model/src/test/java/org/gorpipe/gor/driver/linkfile/UTestLinkFileCacheStats.java b/model/src/test/java/org/gorpipe/gor/driver/linkfile/UTestLinkFileCacheStats.java
new file mode 100644
index 00000000..1d44afbc
--- /dev/null
+++ b/model/src/test/java/org/gorpipe/gor/driver/linkfile/UTestLinkFileCacheStats.java
@@ -0,0 +1,159 @@
+package org.gorpipe.gor.driver.linkfile;
+
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.contrib.java.lang.system.RestoreSystemProperties;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests for the link resolution tally behind ENGKNOW-3770.
+ *
+ * The numbers this produces decide two things: whether caching link content is worth anything
+ * (repeat resolutions of one path), and whether caching *versioned* links would be safe (how often the
+ * content of one path actually changes between resolutions). Simple links are cached, so their repeat
+ * resolutions show up as hits; versioned links are always re-read, so theirs show up as reads.
+ */
+public class UTestLinkFileCacheStats {
+
+ private static final String SIMPLE = "source/data.gorz\n";
+ private static final String SIMPLE_REWRITTEN = "source/other.gorz\n";
+
+ private static final String VERSIONED = """
+ ## SERIAL = 1
+ ## VERSION = 1
+ #FILE\tTIMESTAMP\tMD5\tSERIAL\tINFO
+ source/versions/generation_1.gorz\t2026-01-01T00:00:00.000Z\tMD5SUM1\t1\t
+ """;
+
+ @Rule
+ public final RestoreSystemProperties restoreSystemProperties = new RestoreSystemProperties();
+
+ @Before
+ public void setUp() {
+ System.setProperty("gor.driver.link.cache.stats", "true");
+ LinkFileCacheStats.reset();
+ }
+
+ private static void enable(boolean enabled) {
+ System.setProperty("gor.driver.link.cache.stats", String.valueOf(enabled));
+ }
+
+ /**
+ * A link file gets converted from simple to versioned in normal operation (LinkUpdateCommand does
+ * exactly this). Reads follow the conversion because they derive the version from the content they
+ * just read; hits must follow it too, or the per-version hit rate -- the number this tally exists to
+ * produce -- is skewed in both directions.
+ */
+ @Test
+ public void hitsFollowThePathsCurrentVersion() {
+ var path = "s3://thebucket/ref/dbsnp.gorz.link";
+
+ LinkFileCacheStats.recordRead(path, SIMPLE);
+ LinkFileCacheStats.recordRead(path, VERSIONED);
+ LinkFileCacheStats.recordCacheHit(path);
+
+ assertEquals("a hit must count against the version the path holds now",
+ 1, LinkFileCacheStats.summary("1").hits());
+ assertEquals(0, LinkFileCacheStats.summary("0").hits());
+ assertEquals(1, LinkFileCacheStats.summary("1").distinctPaths());
+ assertEquals(0, LinkFileCacheStats.summary("0").distinctPaths());
+ }
+
+ /**
+ * The first stats-enabled resolution registers a shutdown hook to dump the tally. If that first
+ * resolution happens while the JVM is already shutting down, registering throws -- and an
+ * observe-only tally must never be able to fail the read it is counting.
+ */
+ @Test
+ public void readsSurviveAJvmThatIsAlreadyShuttingDown() {
+ LinkFileCacheStats.shutdownHookRegistrar = hook -> {
+ throw new IllegalStateException("Shutdown in progress");
+ };
+
+ LinkFileCacheStats.recordRead("s3://thebucket/ref/dbsnp.gorz.link", SIMPLE);
+ LinkFileCacheStats.recordCacheHit("s3://thebucket/ref/dbsnp.gorz.link");
+
+ assertEquals(1, LinkFileCacheStats.summary("0").reads());
+ assertEquals(1, LinkFileCacheStats.summary("0").hits());
+ }
+
+ @Test
+ public void recordsNothingWhenDisabled() {
+ enable(false);
+
+ LinkFileCacheStats.recordRead("s3://thebucket/ref/dbsnp.gorz.link", SIMPLE);
+ LinkFileCacheStats.recordCacheHit("s3://thebucket/ref/dbsnp.gorz.link");
+
+ assertEquals("must cost nothing unless explicitly switched on",
+ 0, LinkFileCacheStats.summary("0").reads());
+ assertEquals(0, LinkFileCacheStats.summary("0").hits());
+ }
+
+ @Test
+ public void repeatReadsOfOnePathCountAsOneDistinctPath() {
+ var path = "s3://thebucket/ref/dbsnp.gorz.link";
+
+ LinkFileCacheStats.recordRead(path, SIMPLE);
+ LinkFileCacheStats.recordRead(path, SIMPLE);
+ LinkFileCacheStats.recordRead(path, SIMPLE);
+
+ var simple = LinkFileCacheStats.summary("0");
+ assertEquals(3, simple.reads());
+ assertEquals("repeat resolutions of one path are what a cache would save", 1, simple.distinctPaths());
+ assertEquals("unchanged content is not a change", 0, simple.contentChanges());
+ }
+
+ @Test
+ public void changedContentOnOnePathIsCountedAsAChange() {
+ var path = "s3://thebucket/ref/dbsnp.gorz.link";
+
+ LinkFileCacheStats.recordRead(path, SIMPLE);
+ LinkFileCacheStats.recordRead(path, SIMPLE_REWRITTEN);
+ LinkFileCacheStats.recordRead(path, SIMPLE_REWRITTEN);
+
+ assertEquals("only the resolution that saw new content counts",
+ 1, LinkFileCacheStats.summary("0").contentChanges());
+ }
+
+ @Test
+ public void simpleAndVersionedAreTalliedSeparately() {
+ LinkFileCacheStats.recordRead("s3://thebucket/ref/simple.gorz.link", SIMPLE);
+ LinkFileCacheStats.recordRead("s3://thebucket/ref/versioned.gorz.link", VERSIONED);
+ LinkFileCacheStats.recordRead("s3://thebucket/ref/versioned.gorz.link", VERSIONED);
+
+ assertEquals("a bare link path is a simple link file", 1, LinkFileCacheStats.summary("0").reads());
+ assertEquals("a '## VERSION = 1' header is a versioned link file",
+ 2, LinkFileCacheStats.summary("1").reads());
+ assertEquals(1, LinkFileCacheStats.summary("1").distinctPaths());
+ }
+
+ @Test
+ public void cacheHitsAreCountedAgainstSimpleLinks() {
+ var path = "s3://thebucket/ref/dbsnp.gorz.link";
+
+ LinkFileCacheStats.recordRead(path, SIMPLE);
+ LinkFileCacheStats.recordCacheHit(path);
+ LinkFileCacheStats.recordCacheHit(path);
+
+ var simple = LinkFileCacheStats.summary("0");
+ assertEquals(1, simple.reads());
+ assertEquals("only simple links are cached, so hits belong to them", 2, simple.hits());
+ }
+
+ @Test
+ public void distinctPathsAreCappedSoTheTallyCannotGrowWithoutBound() {
+ System.setProperty("gor.driver.link.cache.stats.maxpaths", "2");
+
+ for (int i = 0; i < 10; i++) {
+ LinkFileCacheStats.recordRead("s3://thebucket/ref/file" + i + ".gorz.link", SIMPLE);
+ }
+
+ var simple = LinkFileCacheStats.summary("0");
+ assertTrue("tracked paths must stay within the cap, got " + simple.distinctPaths(),
+ simple.distinctPaths() <= 2);
+ assertEquals("every resolution is still counted", 10, simple.reads());
+ }
+}
diff --git a/model/src/test/java/org/gorpipe/gor/session/UTestGorSessionCache.java b/model/src/test/java/org/gorpipe/gor/session/UTestGorSessionCache.java
index df7bdf94..4881e330 100644
--- a/model/src/test/java/org/gorpipe/gor/session/UTestGorSessionCache.java
+++ b/model/src/test/java/org/gorpipe/gor/session/UTestGorSessionCache.java
@@ -1,6 +1,6 @@
package org.gorpipe.gor.session;
-import org.gorpipe.gor.driver.providers.stream.sources.StreamSource;
+import org.gorpipe.gor.driver.linkfile.CachedLinkContent;
import org.junit.Test;
import java.time.Duration;
@@ -8,7 +8,6 @@
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
-import static org.mockito.Mockito.mock;
/**
* Tests for the caches held by a {@link GorSessionCache}.
@@ -49,14 +48,15 @@ public void s3MetadataCacheEntriesExpire() {
*/
@Test
public void linkContentIsNotSharedBetweenSessions() {
- var source = mock(StreamSource.class);
+ var linkPath = "s3://thebucket/ref/dbsnp.gorz.link";
var readingSession = new GorSessionCache();
var otherSession = new GorSessionCache();
- readingSession.getLinkCache().put(source, "source/versions/generation_1200.gorz");
+ readingSession.getLinkCache().put(linkPath,
+ new CachedLinkContent("source/versions/generation_1200.gorz", System.nanoTime()));
assertNull("link content cached by one session must not be visible to another",
- otherSession.getLinkCache().getIfPresent(source));
+ otherSession.getLinkCache().getIfPresent(linkPath));
}
@Test
+ *
+ *
+ *