feat(ENGKNOW-3770): cache simple link files, instrument the rest

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.
+ *
+ * <p>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.
+ *
+ * <p>Stamped at the <i>start</i> 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) {
+}

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<StreamSource, String> staticLinkCache = Caffeine.newBuilder()
+    private static final Cache<String, CachedLinkContent> 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<String, Long> 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);
         }
     }
 
@@ -426,32 +441,79 @@ public static String loadContentFromSource(StreamSource source) throws IOExcepti
             return null;
         }
 
-        if (USE_LINK_CACHE) {
-            try {
-                Cache<StreamSource, String> 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();
+        if (linkCache != null) {
+            var cached = linkCache.getIfPresent(path);
+            if (cached != null) {
+                if (isRetiredByWrite(path, cached)) {
+                    linkCache.invalidate(path);
+                } else {
+                    LinkFileCacheStats.recordCacheHit(path);
+                    return cached.content();
                 }
-                throw new IOException(e.getCause());
             }
-        } else {
-            return readLimitedLinkContent(source);
         }
+
+        var readStartedNanos = System.nanoTime();
+        var content = readLimitedLinkContent(source);
+        LinkFileCacheStats.recordRead(path, content);
+
+        if (linkCache != null && isSimpleLinkFile(content)) {
+            linkCache.put(path, new CachedLinkContent(content, readStartedNanos));
+        }
+
+        return content;
+    }
+
+    /** 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.
+     *
+     * <p>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<String, CachedLinkContent> 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.
+     *
+     * <p>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.
+     *
+     * <p>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);
+            }
         }
     }
 

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).
+ *
+ * <p>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:
+ *
+ * <ul>
+ *   <li><b>Is caching worth anything</b> -- resolutions against distinct paths, plus the hits the
+ *       cache actually serves.</li>
+ *   <li><b>Would caching versioned links be safe</b> -- 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.</li>
+ * </ul>
+ *
+ * <p>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<String, Counters> counters = new ConcurrentHashMap<>();
+    // Per path, enough to spot a content change.  Capped.
+    private static final Map<String, TrackedPath> trackedPaths = new ConcurrentHashMap<>();
+
+    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<Thread> shutdownHookRegistrar = hook -> Runtime.getRuntime().addShutdownHook(hook);
+
+    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 {
+        // 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) {
+            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;
+                tracked.version = version;
+                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);
+    }
+
+    /**
+     * 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() {
+        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)) {
+            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);
+            }
+        }
+    }
+}

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<StreamSource, String> linkCache = Caffeine.newBuilder()
+    private final Cache<String, CachedLinkContent> linkCache = Caffeine.newBuilder()
             .maximumSize(10000)
             .expireAfterWrite(5, TimeUnit.MINUTES).build();
 
@@ -107,7 +108,7 @@ public Map<String, Set<String>> getSets() {
     public Cache<String, Object> getS3MetadataCache() {
         return s3MetadataCache;
     }
-    public Cache<StreamSource, String> getLinkCache() {
+    public Cache<String, CachedLinkContent> getLinkCache() {
         return linkCache;
     }