From ce773104915450be8447e43aafba94010e3d6165 Mon Sep 17 00:00:00 2001 From: Gisli Magnusson Date: Wed, 2 Sep 2026 17:10:03 +0000 Subject: [PATCH 1/2] feat(ENGKNOW-3770): cache simple link files, instrument the rest The link content cache could never hit. It was keyed on the StreamSource object identity, no source or wrapper type overrides equals/hashCode, and PluggableGorDriver.getDataSource builds a fresh source for every resolution, so the key was never equal to a previous one. Every link file resolution re-read the link file from storage while gor.driver.link.cache claimed otherwise. Cache simple link files, keyed on the link file path with the existing 5 minute expiry. A simple link file is a bare data path, rewritten so rarely that serving one slightly stale is an accepted trade, and it is by far the common case, so this is where the saving is. Versioned link files are still always re-read. They are rewritten in normal operation by appendEntry and by versioned-link GC, and a stale entry could resolve to a generation GC has already deleted, silently. Keying on last modified and length does not fix that: the metadata is itself served from a cache with its own expiry, so the key would be built from stale values and would still match. - LinkFileCacheStats records, per link file version, reads, cache hits, distinct paths, and how often the content behind one path changed between resolutions. Observe only, off unless gor.driver.link.cache.stats is set, path map capped. The change count for versioned links is what decides whether they can be cached later. - save() invalidates the cached content for its link file. Tolerating another process's rewrite until expiry is the trade; reading back stale content this process just replaced is not. Caught by UTestGorWrite. - An empty link file is never cached. It carries no link and is typically a placeholder about to be written, and caching it made a later read report the wrong version. - readLimitedLinkContent releases the read handle it opened. While the cache was keyed on the source object it pinned every source handed to it, hiding the fact that callers do not close them. Co-Authored-By: Claude Opus 5 (1M context) --- .../gorpipe/gor/driver/linkfile/LinkFile.java | 92 ++++++++--- .../driver/linkfile/LinkFileCacheStats.java | 154 ++++++++++++++++++ .../gorpipe/gor/session/GorSessionCache.java | 4 +- .../driver/linkfile/UTestLinkFileCache.java | 124 ++++++++++++++ .../linkfile/UTestLinkFileCacheStats.java | 120 ++++++++++++++ .../gor/session/UTestGorSessionCache.java | 8 +- 6 files changed, 471 insertions(+), 31 deletions(-) create mode 100644 model/src/main/java/org/gorpipe/gor/driver/linkfile/LinkFileCacheStats.java create mode 100644 model/src/test/java/org/gorpipe/gor/driver/linkfile/UTestLinkFileCache.java create mode 100644 model/src/test/java/org/gorpipe/gor/driver/linkfile/UTestLinkFileCacheStats.java 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..5d0e4517 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,7 @@ 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 staticLinkCache = Caffeine.newBuilder() + private static final Cache staticLinkCache = Caffeine.newBuilder() .maximumSize(10000) .expireAfterWrite(5, TimeUnit.MINUTES).build(); @@ -309,6 +309,10 @@ private void save(OutputStream os, long timestamp, FileReader reader) { validate(); } + // Whatever we cached for this link file is about to be wrong. The cache tolerates a link + // rewritten by another process going unnoticed until it expires, but never one rewritten here. + invalidateCachedContent(source); + meta.setProperty(LinkFileMeta.HEADER_SERIAL_KEY, Integer.toString(Integer.parseInt(meta.getProperty(LinkFileMeta.HEADER_SERIAL_KEY, "0")) + 1)); var currentTimestamp = timestamp > 0 ? timestamp : System.currentTimeMillis(); @@ -426,34 +430,62 @@ public static String loadContentFromSource(StreamSource source) throws IOExcepti return null; } - if (USE_LINK_CACHE) { - try { - Cache linkCache; - if (GorSession.currentSession.get() != null && USE_LINK_CACHE_SESSION) { - linkCache = GorSession.currentSession.get().getCache().getLinkCache(); - } else { - log.warn("No session available, can not use session link cache for " + source); - linkCache = staticLinkCache; - } + var path = source.getFullPath(); + var linkCache = USE_LINK_CACHE ? linkCache(source) : null; - return linkCache.get(source, (k) -> { - try { - return readLimitedLinkContent(k); - } catch (Exception e) { - throw new UncheckedExecutionException(e); - } - }); - } catch (UncheckedExecutionException e) { - if (e.getCause() instanceof IOException) { - throw (IOException) e.getCause(); - } - throw new IOException(e.getCause()); + if (linkCache != null) { + var cached = linkCache.getIfPresent(path); + if (cached != null) { + LinkFileCacheStats.recordCacheHit(path); + return cached; } - } else { - return readLimitedLinkContent(source); + } + + var content = readLimitedLinkContent(source); + LinkFileCacheStats.recordRead(path, content); + + if (linkCache != null && isSimpleLinkFile(content)) { + linkCache.put(path, content); + } + + return content; + } + + /** Drops any cached content for this link file, from both the session and the fallback cache. */ + private static void invalidateCachedContent(StreamSource source) { + var path = source.getFullPath(); + staticLinkCache.invalidate(path); + var session = GorSession.currentSession.get(); + if (session != null) { + session.getCache().getLinkCache().invalidate(path); } } + private static Cache linkCache(StreamSource source) { + if (GorSession.currentSession.get() != null && USE_LINK_CACHE_SESSION) { + return GorSession.currentSession.get().getCache().getLinkCache(); + } + log.warn("No session available, can not use session link cache for " + source); + return staticLinkCache; + } + + /** + * Only simple link files are cached. + * + *

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) { try (InputStream is = source.open()) { var content = StreamUtils.readString(is, 2 * LINK_FILE_MAX_SIZE); @@ -464,6 +496,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..4165722b --- /dev/null +++ b/model/src/main/java/org/gorpipe/gor/driver/linkfile/LinkFileCacheStats.java @@ -0,0 +1,154 @@ +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; + +/** + * 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: + * + *

    + *
  • Is caching worth anything -- resolutions against distinct paths, plus the hits the + * cache actually serves.
    • + *
  • Would caching versioned links be safe -- how often the content behind one path actually + * changes between two resolutions. Versioned links are never cached, so every resolution of one + * is a read and this number is complete for them. Simple links are cached, so once an entry is + * warm their changes are not observed -- which is the accepted trade for that kind of link.
    • + *
+ * + *

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 counters = new ConcurrentHashMap<>(); + // Per path, enough to spot a content change. Capped. + private static final Map trackedPaths = new ConcurrentHashMap<>(); + + private static final AtomicBoolean dumpRegistered = new AtomicBoolean(); + + private LinkFileCacheStats() {} + + /** Tally for one link file version. */ + public record Summary(long reads, long hits, long contentChanges, int distinctPaths) {} + + private static final class Counters { + final AtomicLong reads = new AtomicLong(); + final AtomicLong hits = new AtomicLong(); + final AtomicLong contentChanges = new AtomicLong(); + } + + private static final class TrackedPath { + final String version; + volatile String contentFingerprint; + + TrackedPath(String version, String contentFingerprint) { + this.version = version; + this.contentFingerprint = contentFingerprint; + } + } + + private static boolean enabled() { + return PropsHelper.getBoolean(STATS_ENABLED_KEY, false); + } + + private static int maxPaths() { + return PropsHelper.getInt(MAX_PATHS_KEY, DEFAULT_MAX_PATHS); + } + + private static Counters counters(String version) { + return counters.computeIfAbsent(version, v -> new Counters()); + } + + /** Length plus hash: cheap, and far less collision prone for change detection than the hash alone. */ + private static String fingerprint(String content) { + return content.length() + ":" + content.hashCode(); + } + + static String versionOf(String content) { + return LinkFileMeta.createOrLoad(content, null, false).getVersion(); + } + + /** + * Record a resolution that went to storage. + */ + public static void recordRead(String path, String content) { + if (!enabled() || content == null) { + return; + } + registerDump(); + + var version = versionOf(content); + var counters = counters(version); + counters.reads.incrementAndGet(); + + var fingerprint = fingerprint(content); + var tracked = trackedPaths.get(path); + if (tracked != null) { + if (!tracked.contentFingerprint.equals(fingerprint)) { + tracked.contentFingerprint = fingerprint; + counters.contentChanges.incrementAndGet(); + } + } else if (trackedPaths.size() < maxPaths()) { + trackedPaths.put(path, new TrackedPath(version, fingerprint)); + } + } + + /** + * Record a resolution served from the cache. Only simple link files are cached, so a hit against a + * path we are not tracking is counted against them. + */ + public static void recordCacheHit(String path) { + if (!enabled()) { + return; + } + registerDump(); + + var tracked = trackedPaths.get(path); + counters(tracked != null ? tracked.version : LinkFileV0.VERSION).hits.incrementAndGet(); + } + + public static Summary summary(String version) { + var c = counters.get(version); + var distinct = (int) trackedPaths.values().stream().filter(t -> t.version.equals(version)).count(); + if (c == null) { + return new Summary(0, 0, 0, distinct); + } + return new Summary(c.reads.get(), c.hits.get(), c.contentChanges.get(), distinct); + } + + public static void reset() { + counters.clear(); + trackedPaths.clear(); + } + + public static void logSummary() { + for (var version : counters.keySet().stream().sorted().toList()) { + var s = summary(version); + var resolutions = s.reads() + s.hits(); + log.info("Link file cache stats, version {}: {} resolutions ({} reads, {} cache hits) over {} distinct paths, {} content changes seen", + version, resolutions, s.reads(), s.hits(), s.distinctPaths(), s.contentChanges()); + } + } + + private static void registerDump() { + if (dumpRegistered.compareAndSet(false, true)) { + Runtime.getRuntime().addShutdownHook(new Thread(LinkFileCacheStats::logSummary, + "link-cache-stats-dump")); + } + } +} diff --git a/model/src/main/java/org/gorpipe/gor/session/GorSessionCache.java b/model/src/main/java/org/gorpipe/gor/session/GorSessionCache.java index 45f7e3b6..848f6b43 100644 --- a/model/src/main/java/org/gorpipe/gor/session/GorSessionCache.java +++ b/model/src/main/java/org/gorpipe/gor/session/GorSessionCache.java @@ -59,7 +59,7 @@ public class GorSessionCache { // process, so content read by one session was served to every later one. Link files get rewritten, // so content held past its session may no longer be true. Matches the bounds of the static // fallback cache in LinkFile. - private final Cache linkCache = Caffeine.newBuilder() + private final Cache linkCache = Caffeine.newBuilder() .maximumSize(10000) .expireAfterWrite(5, TimeUnit.MINUTES).build(); @@ -107,7 +107,7 @@ public Map> getSets() { public Cache getS3MetadataCache() { return s3MetadataCache; } - public Cache getLinkCache() { + public Cache getLinkCache() { return linkCache; } diff --git a/model/src/test/java/org/gorpipe/gor/driver/linkfile/UTestLinkFileCache.java b/model/src/test/java/org/gorpipe/gor/driver/linkfile/UTestLinkFileCache.java new file mode 100644 index 00000000..82c62478 --- /dev/null +++ b/model/src/test/java/org/gorpipe/gor/driver/linkfile/UTestLinkFileCache.java @@ -0,0 +1,124 @@ +package org.gorpipe.gor.driver.linkfile; + +import org.gorpipe.gor.driver.providers.stream.sources.file.FileSource; +import org.gorpipe.gor.model.DriverBackedFileReader; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.nio.file.Files; +import java.nio.file.Path; + +import static org.junit.Assert.assertEquals; + +/** + * Tests which link files {@link LinkFile#loadContentFromSource} caches (ENGKNOW-3770). + * + *

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"); + } + + private String read() throws Exception { + return LinkFile.loadContentFromSource(new FileSource(linkPath.toString())); + } + + @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()); + } + + /** 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..c4e18d20 --- /dev/null +++ b/model/src/test/java/org/gorpipe/gor/driver/linkfile/UTestLinkFileCacheStats.java @@ -0,0 +1,120 @@ +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)); + } + + @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..5ebf9ecc 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,5 @@ package org.gorpipe.gor.session; -import org.gorpipe.gor.driver.providers.stream.sources.StreamSource; import org.junit.Test; import java.time.Duration; @@ -8,7 +7,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 +47,14 @@ 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, "source/versions/generation_1200.gorz"); assertNull("link content cached by one session must not be visible to another", - otherSession.getLinkCache().getIfPresent(source)); + otherSession.getLinkCache().getIfPresent(linkPath)); } @Test From 3b3952ac534cb32cce19a228455faadf3f249578 Mon Sep 17 00:00:00 2001 From: Gisli Magnusson Date: Wed, 2 Sep 2026 19:51:36 +0000 Subject: [PATCH 2/2] fix(ENGKNOW-3770): retire cached link content a save cannot reach Review of #138 found three ways the link content cache can keep serving content this process has already replaced. Invalidation ran at the top of save(), before the content was written and, for an atomic write, before close() renamed the temp file into place. A resolution in that window read the pre-write content and re-cached it, so the stale link was served for the full 5 minute expiry. It now runs in a finally after the stream is closed, so a failed write drops the entry too. That alone is not enough. Link content is cached per session, and GorSession.currentSession is an InheritableThreadLocal that nothing clears, so a read and a save of the same link file can land on different sessions: the save can only invalidate the cache bound to its own thread, and a resolution already in flight can put its pre-write content into another session's cache after the invalidate. Instead of trying to reach every cache, a save now records the time of the write, and a cached entry whose read started before that mark is dropped on lookup. Entries carry the time their read started, not the time they were cached: a read that opened its stream before the file was replaced holds the old content even though it finished afterwards. The marks are bounded and expire like the content caches, so a mark outlives every entry it has to retire. Two fixes in the resolution tally, which must never affect a read: - Registering the shutdown hook that dumps the tally threw IllegalStateException if the first stats-enabled resolution happened while the JVM was already shutting down, failing the read it was counting. - TrackedPath.version was fixed at first sight, so after a link was converted from simple to versioned its reads moved to the new version while its hits stayed on the old one, skewing the per-version hit rate in both directions. Co-Authored-By: Claude Opus 5 (1M context) --- .../driver/linkfile/CachedLinkContent.java | 20 ++++ .../gorpipe/gor/driver/linkfile/LinkFile.java | 50 ++++++++-- .../driver/linkfile/LinkFileCacheStats.java | 26 +++++- .../gorpipe/gor/session/GorSessionCache.java | 5 +- .../driver/linkfile/UTestLinkFileCache.java | 91 +++++++++++++++++++ .../linkfile/UTestLinkFileCacheStats.java | 39 ++++++++ .../gor/session/UTestGorSessionCache.java | 4 +- 7 files changed, 219 insertions(+), 16 deletions(-) create mode 100644 model/src/main/java/org/gorpipe/gor/driver/linkfile/CachedLinkContent.java diff --git a/model/src/main/java/org/gorpipe/gor/driver/linkfile/CachedLinkContent.java b/model/src/main/java/org/gorpipe/gor/driver/linkfile/CachedLinkContent.java new file mode 100644 index 00000000..21b24a6e --- /dev/null +++ b/model/src/main/java/org/gorpipe/gor/driver/linkfile/CachedLinkContent.java @@ -0,0 +1,20 @@ +package org.gorpipe.gor.driver.linkfile; + +/** + * Link file content held in a cache, stamped with when the read that produced it started. + * + *

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 5d0e4517..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 staticLinkCache = Caffeine.newBuilder() + private static final Cache staticLinkCache = Caffeine.newBuilder() + .maximumSize(10000) + .expireAfterWrite(5, TimeUnit.MINUTES).build(); + + // A save cannot reach the link caches of other sessions, so it leaves a mark here instead and any + // cached entry for the path whose read started before the mark is dropped on lookup. Bounded and + // expiring like the content caches, so a mark outlives every entry it has to retire. + private static final Cache linkWriteMarks = Caffeine.newBuilder() .maximumSize(10000) .expireAfterWrite(5, TimeUnit.MINUTES).build(); @@ -300,6 +307,14 @@ public void save(long timestamp, FileReader reader) { save(os, timestamp, reader); } catch (IOException e) { throw new GorResourceException("Could not save: " + source.getFullPath(), source.getFullPath(), e); + } finally { + // Whatever we cached for this link file is now wrong. The cache tolerates a link rewritten + // by another process going unnoticed until it expires, but never one rewritten here. + // + // After the stream is closed, not before: closing is what renames the temp file into place + // for an atomic write, so a resolution running in between would otherwise read the old + // content and re-cache it behind us. In a finally so a failed write drops it too. + invalidateCachedContent(source); } } @@ -309,10 +324,6 @@ private void save(OutputStream os, long timestamp, FileReader reader) { validate(); } - // Whatever we cached for this link file is about to be wrong. The cache tolerates a link - // rewritten by another process going unnoticed until it expires, but never one rewritten here. - invalidateCachedContent(source); - meta.setProperty(LinkFileMeta.HEADER_SERIAL_KEY, Integer.toString(Integer.parseInt(meta.getProperty(LinkFileMeta.HEADER_SERIAL_KEY, "0")) + 1)); var currentTimestamp = timestamp > 0 ? timestamp : System.currentTimeMillis(); @@ -436,24 +447,43 @@ public static String loadContentFromSource(StreamSource source) throws IOExcepti if (linkCache != null) { var cached = linkCache.getIfPresent(path); if (cached != null) { - LinkFileCacheStats.recordCacheHit(path); - return cached; + if (isRetiredByWrite(path, cached)) { + linkCache.invalidate(path); + } else { + LinkFileCacheStats.recordCacheHit(path); + return cached.content(); + } } } + var readStartedNanos = System.nanoTime(); var content = readLimitedLinkContent(source); LinkFileCacheStats.recordRead(path, content); if (linkCache != null && isSimpleLinkFile(content)) { - linkCache.put(path, content); + linkCache.put(path, new CachedLinkContent(content, readStartedNanos)); } return content; } - /** Drops any cached content for this link file, from both the session and the fallback cache. */ + /** True if a save of this link file completed after the read behind this entry began. */ + private static boolean isRetiredByWrite(String path, CachedLinkContent cached) { + var writtenAtNanos = linkWriteMarks.getIfPresent(path); + return writtenAtNanos != null && cached.readStartedNanos() - writtenAtNanos <= 0; + } + + /** + * Retires every cached copy of this link file's content, in this session and in any other. + * + *

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) { @@ -461,7 +491,7 @@ private static void invalidateCachedContent(StreamSource source) { } } - private static Cache linkCache(StreamSource source) { + private static Cache linkCache(StreamSource source) { if (GorSession.currentSession.get() != null && USE_LINK_CACHE_SESSION) { return GorSession.currentSession.get().getCache().getLinkCache(); } 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 index 4165722b..79c6f6b8 100644 --- a/model/src/main/java/org/gorpipe/gor/driver/linkfile/LinkFileCacheStats.java +++ b/model/src/main/java/org/gorpipe/gor/driver/linkfile/LinkFileCacheStats.java @@ -6,6 +6,7 @@ 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). @@ -41,6 +42,10 @@ public final class LinkFileCacheStats { private static final AtomicBoolean dumpRegistered = new AtomicBoolean(); + // Package private and not final so a test can make registration fail the way an already shutting + // down JVM does. + static Consumer shutdownHookRegistrar = hook -> Runtime.getRuntime().addShutdownHook(hook); + private LinkFileCacheStats() {} /** Tally for one link file version. */ @@ -53,7 +58,9 @@ private static final class Counters { } private static final class TrackedPath { - final String version; + // Not final: a link file gets converted from simple to versioned in normal operation, and the + // version has to follow the content or hits and reads for the path land in different buckets. + volatile String version; volatile String contentFingerprint; TrackedPath(String version, String contentFingerprint) { @@ -101,6 +108,7 @@ public static void recordRead(String path, String content) { if (tracked != null) { if (!tracked.contentFingerprint.equals(fingerprint)) { tracked.contentFingerprint = fingerprint; + tracked.version = version; counters.contentChanges.incrementAndGet(); } } else if (trackedPaths.size() < maxPaths()) { @@ -131,9 +139,15 @@ public static Summary summary(String version) { return new Summary(c.reads.get(), c.hits.get(), c.contentChanges.get(), distinct); } + /** + * Clears the tally. A test hook: it also clears the shutdown hook registration, and leaves the + * registrar a no-op so repeated resets in one JVM do not pile up dump hooks. + */ public static void reset() { counters.clear(); trackedPaths.clear(); + dumpRegistered.set(false); + shutdownHookRegistrar = hook -> { }; } public static void logSummary() { @@ -147,8 +161,14 @@ public static void logSummary() { private static void registerDump() { if (dumpRegistered.compareAndSet(false, true)) { - Runtime.getRuntime().addShutdownHook(new Thread(LinkFileCacheStats::logSummary, - "link-cache-stats-dump")); + try { + shutdownHookRegistrar.accept(new Thread(LinkFileCacheStats::logSummary, + "link-cache-stats-dump")); + } catch (IllegalStateException e) { + // The JVM is already shutting down, so there is no hook to run and nothing to do about + // it. This tally only observes -- it must never be able to fail the read it counts. + log.debug("Link file cache stats will not be dumped, the JVM is already shutting down", e); + } } } } diff --git a/model/src/main/java/org/gorpipe/gor/session/GorSessionCache.java b/model/src/main/java/org/gorpipe/gor/session/GorSessionCache.java index 848f6b43..93bab8b9 100644 --- a/model/src/main/java/org/gorpipe/gor/session/GorSessionCache.java +++ b/model/src/main/java/org/gorpipe/gor/session/GorSessionCache.java @@ -24,6 +24,7 @@ import com.github.benmanes.caffeine.cache.Cache; import com.github.benmanes.caffeine.cache.Caffeine; +import org.gorpipe.gor.driver.linkfile.CachedLinkContent; import org.gorpipe.gor.driver.providers.stream.sources.StreamSource; import org.gorpipe.util.Pair; @@ -59,7 +60,7 @@ public class GorSessionCache { // process, so content read by one session was served to every later one. Link files get rewritten, // so content held past its session may no longer be true. Matches the bounds of the static // fallback cache in LinkFile. - private final Cache linkCache = Caffeine.newBuilder() + private final Cache linkCache = Caffeine.newBuilder() .maximumSize(10000) .expireAfterWrite(5, TimeUnit.MINUTES).build(); @@ -107,7 +108,7 @@ public Map> getSets() { public Cache getS3MetadataCache() { return s3MetadataCache; } - public Cache getLinkCache() { + public Cache getLinkCache() { return linkCache; } diff --git a/model/src/test/java/org/gorpipe/gor/driver/linkfile/UTestLinkFileCache.java b/model/src/test/java/org/gorpipe/gor/driver/linkfile/UTestLinkFileCache.java index 82c62478..b5fb68bf 100644 --- a/model/src/test/java/org/gorpipe/gor/driver/linkfile/UTestLinkFileCache.java +++ b/model/src/test/java/org/gorpipe/gor/driver/linkfile/UTestLinkFileCache.java @@ -2,13 +2,20 @@ import org.gorpipe.gor.driver.providers.stream.sources.file.FileSource; import org.gorpipe.gor.model.DriverBackedFileReader; +import org.gorpipe.gor.session.GorSession; +import org.gorpipe.gor.session.GorSessionCache; +import org.junit.After; import org.junit.Before; import org.junit.Rule; import org.junit.Test; import org.junit.rules.TemporaryFolder; +import java.io.IOException; +import java.io.InputStream; +import java.io.UncheckedIOException; import java.nio.file.Files; import java.nio.file.Path; +import java.util.concurrent.atomic.AtomicBoolean; import static org.junit.Assert.assertEquals; @@ -56,10 +63,41 @@ 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); @@ -100,6 +138,59 @@ public void savingALinkFileInvalidatesItsCachedContent() throws Exception { 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 { 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 index c4e18d20..1d44afbc 100644 --- a/model/src/test/java/org/gorpipe/gor/driver/linkfile/UTestLinkFileCacheStats.java +++ b/model/src/test/java/org/gorpipe/gor/driver/linkfile/UTestLinkFileCacheStats.java @@ -41,6 +41,45 @@ 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); 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 5ebf9ecc..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,5 +1,6 @@ package org.gorpipe.gor.session; +import org.gorpipe.gor.driver.linkfile.CachedLinkContent; import org.junit.Test; import java.time.Duration; @@ -51,7 +52,8 @@ public void linkContentIsNotSharedBetweenSessions() { var readingSession = new GorSessionCache(); var otherSession = new GorSessionCache(); - readingSession.getLinkCache().put(linkPath, "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(linkPath));