29227150127

Committed 13 Jul 2026 05:46AM UTC coverage: 28.504% (+0.02%) from 28.481%

Build # 29227150127

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #552 from knowledgepixels/fix/apicache-replaced-eviction-npe

fix: prevent ApiCache metadata wipe on entry replacement causing home page hang

Coverage Stats

1872 of 7381 branches covered (25.36%)

Branch coverage included in aggregate %.

3777 of 12437 relevant lines covered (30.37%)

4.52 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

39.74

src/main/java/com/knowledgepixels/nanodash/ApiCache.java

package com.knowledgepixels.nanodash;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.RemovalCause;
import com.google.common.cache.RemovalNotification;
import org.apache.wicket.MetaDataKey;
import org.apache.wicket.request.cycle.RequestCycle;
import org.eclipse.rdf4j.model.Model;
import org.nanopub.extra.services.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;

/**
 * A utility class for caching API responses and maps to reduce redundant API calls.
 * This class is thread-safe and ensures that cached data is refreshed periodically.
 */
public class ApiCache {

    private ApiCache() {
    } // no instances allowed

    private static final int MAX_CACHE_ENTRIES = 10_000;

    // How stale a cached response may be before the next access triggers a
    // background re-fetch. Must stay reasonably high: every render that finds a
    // query older than this submits a refresh to the shared pool, which uses a
    // CallerRunsPolicy — so too low a value turns page renders into a refresh
    // storm that can run queries synchronously on the request thread.
    private static final long REFRESH_AGE_THRESHOLD_MS = 60 * 1000;

    // How long a cached response is still served immediately (while refreshing in
    // the background) before it is treated as absent and the caller waits for a
    // fresh fetch. Acts as the stale-data fallback during API outages.
    private static final long MAX_CACHE_AGE_MS = 24 * 60 * 60 * 1000;

    // Upper bound a synchronous caller waits for an in-flight refresh started by
    // another thread when it has nothing cached yet. Without this wait the caller
    // returns null, letting repositories memoise an empty snapshot (see
    // retrieveResponseSync).
    private static final long SYNC_WAIT_FOR_INFLIGHT_MS = 10 * 1000;

    private static final Cache<String, ApiResponse> cachedResponses = CacheBuilder.newBuilder()
        .maximumSize(MAX_CACHE_ENTRIES)
        .expireAfterAccess(24, TimeUnit.HOURS)
        .removalListener(ApiCache::cleanupMetadataOnRemoval)
        .build();
    private static final Cache<String, Model> cachedRdfModels = CacheBuilder.newBuilder()
        .maximumSize(MAX_CACHE_ENTRIES)
        .expireAfterAccess(24, TimeUnit.HOURS)
        .removalListener(ApiCache::cleanupMetadataOnRemoval)
        .build();
    private transient static ConcurrentMap<String, Integer> failed = new ConcurrentHashMap<>();
    private static final Cache<String, Map<String, String>> cachedMaps = CacheBuilder.newBuilder()
        .maximumSize(MAX_CACHE_ENTRIES)
        .expireAfterAccess(24, TimeUnit.HOURS)
        .removalListener(ApiCache::cleanupMetadataOnRemoval)
        .build();
    private transient static ConcurrentMap<String, Long> lastRefresh = new ConcurrentHashMap<>();
    private transient static ConcurrentMap<String, Long> refreshStart = new ConcurrentHashMap<>();
    private transient static ConcurrentMap<String, Long> runAfter = new ConcurrentHashMap<>();
    private static final Logger logger = LoggerFactory.getLogger(ApiCache.class);

    // Guava fires removal notifications also when an entry is REPLACED (every routine
    // refresh's put), and processes them lazily during later cache operations. Cleaning
    // up on a replacement would wipe the metadata of a still-cached entry — in
    // particular a missing lastRefresh timestamp used to make retrieveResponseSync
    // throw an NPE on every call until the entry expired.
    private static void cleanupMetadataOnRemoval(RemovalNotification<String, ?> n) {
        if (n.getCause() == RemovalCause.REPLACED) return;
        cleanupMetadata(n.getKey());
    }

    private static void cleanupMetadata(String cacheId) {
        lastRefresh.remove(cacheId);
        failed.remove(cacheId);
        runAfter.remove(cacheId);
    }

    /**
     * Checks if a cache refresh is currently running for the given cache ID.
     *
     * @param cacheId The unique identifier for the cache.
     * @return True if a refresh is running, false otherwise.
     */
    private static boolean isRunning(String cacheId) {
        Long start = refreshStart.get(cacheId);
        if (start == null) return false;
        return System.currentTimeMillis() - start < 60 * 1000;
    }

    /**
     * Checks if a cache refresh is currently running for the given QueryRef.
     *
     * @param queryRef The query reference
     * @return True if a refresh is running, false otherwise.
     */
    public static boolean isRunning(QueryRef queryRef) {
        return isRunning(queryRef.getAsUrlString());
    }

    /**
     * Request-scoped flag set by {@code NanodashPage} when the current request is a
     * genuine browser reload (the browser sends {@code Cache-Control: max-age=0} or
     * {@code no-cache}). When set, the first access to each query during the page
     * render evicts that query's cache so it re-fetches fresh, while normal
     * navigation, Ajax updates, and the auto-refresh redirect keep serving the
     * cache. Public so the page layer can set it.
     */
    public static final MetaDataKey<Boolean> FORCE_REFRESH_ON_RELOAD = new MetaDataKey<>() {};

    // The query cache-ids already force-evicted during the current reload request,
    // so each is evicted only once (the lazy-load that follows must not re-evict).
    private static final MetaDataKey<HashSet<String>> RELOAD_FORCED_IDS = new MetaDataKey<>() {};

    /**
     * On a genuine browser reload, returns true the first time a given query is
     * accessed this request (and records it), so callers evict its cache once.
     * Returns false on non-reload requests, off the request thread, and for any
     * query already handled this request — so it never triggers a refresh storm.
     */
    private static boolean isForcedReload(String cacheId) {
        RequestCycle rc = RequestCycle.get();
        if (rc == null) return false;
        Boolean force = rc.getMetaData(FORCE_REFRESH_ON_RELOAD);
        if (force == null || !force) return false;
        HashSet<String> handled = rc.getMetaData(RELOAD_FORCED_IDS);
        if (handled == null) {
            handled = new HashSet<>();
            rc.setMetaData(RELOAD_FORCED_IDS, handled);
        }
        return handled.add(cacheId);
    }

    /**
     * Updates the cached API response for a specific query reference.
     *
     * @param queryRef The query reference
     * @throws FailedApiCallException If the API call fails.
     */
    private static void updateResponse(QueryRef queryRef, boolean forced) throws FailedApiCallException, APINotReachableException, NotEnoughAPIInstancesException {
        ApiResponse response;
        if (forced) {
            response = QueryApiAccess.forcedGet(queryRef);
        } else {
            response = QueryApiAccess.get(queryRef);
        }
        String cacheId = queryRef.getAsUrlString();
        logger.info("Updating cached API response for {}", cacheId);
        cachedResponses.put(cacheId, response);
        lastRefresh.put(cacheId, System.currentTimeMillis());
    }

    public static ApiResponse retrieveResponseSync(QueryRef queryRef, boolean forced) {
        long timeNow = System.currentTimeMillis();
        String cacheId = queryRef.getAsUrlString();
        logger.debug("Retrieving cached API response synchronously for {}", cacheId);
        boolean needsRefresh = true;
        if (cachedResponses.getIfPresent(cacheId) != null) {
            // lastRefresh can be missing for a cached entry (racing invalidation or
            // refresh); treat that as stale rather than NPEing on the unboxing.
            Long lastRefreshTime = lastRefresh.get(cacheId);
            needsRefresh = lastRefreshTime == null || timeNow - lastRefreshTime > REFRESH_AGE_THRESHOLD_MS;
        }
        Integer failedCount = failed.get(cacheId);
        if (failedCount != null && failedCount > 2) {
            failed.remove(cacheId);
            throw new RuntimeException("Query failed: " + cacheId);
        }
        if ((needsRefresh || forced) && !isRunning(cacheId)) {
            logger.info("Refreshing cache for {}", cacheId);
            refreshStart.put(cacheId, timeNow);
            try {
                Long after = runAfter.get(cacheId);
                if (after != null) {
                    while (System.currentTimeMillis() < after) {
                        Thread.sleep(100);
                    }
                    runAfter.remove(cacheId);
                }
                if (failed.get(cacheId) != null) {
                    // 1 second pause between failed attempts;
                    Thread.sleep(1000);
                }
                Thread.sleep(100 + new Random().nextLong(400));
            } catch (InterruptedException ex) {
                logger.error("Interrupted while waiting to refresh cache: {}", ex.getMessage());
            }
            try {
                ApiCache.updateResponse(queryRef, forced);
                failed.remove(cacheId);
            } catch (Exception ex) {
                logger.error("Failed to update cache for {}: {}", cacheId, ex.getMessage());
                // Keep stale cached data if available, only invalidate if nothing was cached
                if (cachedResponses.getIfPresent(cacheId) == null) {
                    failed.merge(cacheId, 1, Integer::sum);
                }
                lastRefresh.put(cacheId, System.currentTimeMillis());
            } finally {
                refreshStart.remove(cacheId);
            }
        } else if (cachedResponses.getIfPresent(cacheId) == null && isRunning(cacheId)) {
            // Another thread is doing the first fetch of this query and we have
            // nothing cached yet. Wait for it rather than returning null: a null
            // here lets a caller (e.g. SpaceRepository) memoise an EMPTY snapshot,
            // which then poisons MaintainedResourceRepository.build() and breaks
            // the home page until the next refresh. This adds no new work; it only
            // waits on the refresh already in flight.
            try {
                long deadline = timeNow + SYNC_WAIT_FOR_INFLIGHT_MS;
                while (isRunning(cacheId)
                        && cachedResponses.getIfPresent(cacheId) == null
                        && System.currentTimeMillis() < deadline) {
                    Thread.sleep(50);
                }
            } catch (InterruptedException ex) {
                Thread.currentThread().interrupt();
            }
        }
        return cachedResponses.getIfPresent(cacheId);
    }

    /**
     * Retrieves a cached API response for a specific QueryRef.
     *
     * @param queryRef The QueryRef object containing the query name and parameters.
     * @return The cached API response, or null if not cached.
     */
    public static ApiResponse retrieveResponseAsync(QueryRef queryRef) {
        long timeNow = System.currentTimeMillis();
        String cacheId = queryRef.getAsUrlString();
        logger.debug("Retrieving cached API response asynchronously for {}", cacheId);
        if (isForcedReload(cacheId)) {
            cachedResponses.invalidate(cacheId);
            lastRefresh.remove(cacheId);
        }
        boolean isCached = false;
        boolean needsRefresh = true;
        if (cachedResponses.getIfPresent(cacheId) != null) {
            Long lastRefreshTime = lastRefresh.get(cacheId);
            isCached = lastRefreshTime != null && timeNow - lastRefreshTime < MAX_CACHE_AGE_MS;
            needsRefresh = lastRefreshTime == null || timeNow - lastRefreshTime > REFRESH_AGE_THRESHOLD_MS;
        }
        Integer failedCount = failed.get(cacheId);
        if (failedCount != null && failedCount > 2) {
            failed.remove(cacheId);
            throw new RuntimeException("Query failed: " + cacheId);
        }
        if (needsRefresh && !isRunning(cacheId)) {
            NanodashThreadPool.submit(() -> {
                refreshStart.put(cacheId, System.currentTimeMillis());
                try {
                    Long after = runAfter.get(cacheId);
                    if (after != null) {
                        while (System.currentTimeMillis() < after) {
                            Thread.sleep(100);
                        }
                        runAfter.remove(cacheId);
                    }
                    if (failed.get(cacheId) != null) {
                        // 1 second pause between failed attempts;
                        Thread.sleep(1000);
                    }
                    Thread.sleep(100 + new Random().nextLong(400));
                } catch (InterruptedException ex) {
                    logger.error("Interrupted while waiting to refresh cache: {}", ex.getMessage());
                }
                try {
                    ApiCache.updateResponse(queryRef, false);
                    failed.remove(cacheId);
                } catch (Exception ex) {
                    logger.error("Failed to update cache for {}: {}", cacheId, ex.getMessage());
                    if (cachedResponses.getIfPresent(cacheId) == null) {
                        failed.merge(cacheId, 1, Integer::sum);
                    }
                    lastRefresh.put(cacheId, System.currentTimeMillis());
                } finally {
                    refreshStart.remove(cacheId);
                }
            });
        }
        if (isCached) {
            return cachedResponses.getIfPresent(cacheId);
        } else {
            return null;
        }
    }

    /**
     * Updates the cached map for a specific query reference.
     *
     * @param queryRef The query reference
     * @throws FailedApiCallException If the API call fails.
     */
    private static void updateMap(QueryRef queryRef) throws FailedApiCallException, APINotReachableException, NotEnoughAPIInstancesException {
        Map<String, String> map = new HashMap<>();
        List<ApiResponseEntry> respList = QueryApiAccess.get(queryRef).getData();
        while (respList != null && !respList.isEmpty()) {
            ApiResponseEntry resultEntry = respList.removeFirst();
            map.put(resultEntry.get("key"), resultEntry.get("value"));
        }
        String cacheId = queryRef.getAsUrlString();
        cachedMaps.put(cacheId, map);
        lastRefresh.put(cacheId, System.currentTimeMillis());
    }

    /**
     * Retrieves a cached map for a specific query reference.
     * If the cache is stale, it triggers a background refresh.
     *
     * @param queryRef The query reference
     * @return The cached map, or null if not cached.
     */
    public static Map<String, String> retrieveMap(QueryRef queryRef) {
        long timeNow = System.currentTimeMillis();
        String cacheId = queryRef.getAsUrlString();
        if (isForcedReload(cacheId)) {
            cachedMaps.invalidate(cacheId);
            lastRefresh.remove(cacheId);
        }
        boolean isCached = false;
        boolean needsRefresh = true;
        if (cachedMaps.getIfPresent(cacheId) != null) {
            Long lastRefreshTime = lastRefresh.get(cacheId);
            isCached = lastRefreshTime != null && timeNow - lastRefreshTime < MAX_CACHE_AGE_MS;
            needsRefresh = lastRefreshTime == null || timeNow - lastRefreshTime > REFRESH_AGE_THRESHOLD_MS;
        }
        if (needsRefresh && !isRunning(cacheId)) {
            NanodashThreadPool.submit(() -> {
                refreshStart.put(cacheId, System.currentTimeMillis());
                try {
                    Long after = runAfter.get(cacheId);
                    if (after != null) {
                        while (System.currentTimeMillis() < after) {
                            Thread.sleep(100);
                        }
                        runAfter.remove(cacheId);
                    }
                    Thread.sleep(100 + new Random().nextLong(400));
                } catch (InterruptedException ex) {
                    logger.error("Interrupted while waiting to refresh cache: {}", ex.getMessage());
                }
                try {
                    ApiCache.updateMap(queryRef);
                } catch (Exception ex) {
                    logger.error("Failed to update cache for {}: {}", cacheId, ex.getMessage());
                    cachedMaps.invalidate(cacheId);
                    lastRefresh.put(cacheId, System.currentTimeMillis());
                }  finally {
                    refreshStart.remove(cacheId);
                }
            });
        }
        if (isCached) {
            return cachedMaps.getIfPresent(cacheId);
        } else {
            return null;
        }
    }

    private static void updateRdfModel(QueryRef queryRef) throws FailedApiCallException, APINotReachableException, NotEnoughAPIInstancesException {
        final Model[] modelRef = new Model[1];
        QueryAccess qa = new QueryAccess() {
            @Override
            protected void processHeader(String[] line) {}
            @Override
            protected void processLine(String[] line) {}
            @Override
            protected void processRdfContent(Model model) {
                modelRef[0] = model;
            }
        };
        qa.call(queryRef);
        if (modelRef[0] == null) {
            throw new FailedApiCallException(new Exception("No RDF content in response for query: " + queryRef.getQueryId()));
        }
        String cacheId = queryRef.getAsUrlString();
        logger.info("Updating cached RDF model for {}", cacheId);
        cachedRdfModels.put(cacheId, modelRef[0]);
        lastRefresh.put(cacheId, System.currentTimeMillis());
    }

    /**
     * Retrieves a cached RDF model for a CONSTRUCT query, triggering a background fetch if needed.
     *
     * @param queryRef The QueryRef for the CONSTRUCT query.
     * @return The cached RDF Model, or null if not yet available.
     */
    public static Model retrieveRdfModelAsync(QueryRef queryRef) {
        long timeNow = System.currentTimeMillis();
        String cacheId = queryRef.getAsUrlString();
        logger.debug("Retrieving cached RDF model asynchronously for {}", cacheId);
        if (isForcedReload(cacheId)) {
            cachedRdfModels.invalidate(cacheId);
            lastRefresh.remove(cacheId);
        }
        boolean isCached = false;
        boolean needsRefresh = true;
        if (cachedRdfModels.getIfPresent(cacheId) != null) {
            Long lastRefreshTime = lastRefresh.get(cacheId);
            isCached = lastRefreshTime != null && timeNow - lastRefreshTime < MAX_CACHE_AGE_MS;
            needsRefresh = lastRefreshTime == null || timeNow - lastRefreshTime > REFRESH_AGE_THRESHOLD_MS;
        }
        Integer failedCount = failed.get(cacheId);
        if (failedCount != null && failedCount > 2) {
            failed.remove(cacheId);
            throw new RuntimeException("Query failed: " + cacheId);
        }
        if (needsRefresh && !isRunning(cacheId)) {
            NanodashThreadPool.submit(() -> {
                refreshStart.put(cacheId, System.currentTimeMillis());
                try {
                    Long after = runAfter.get(cacheId);
                    if (after != null) {
                        while (System.currentTimeMillis() < after) {
                            Thread.sleep(100);
                        }
                        runAfter.remove(cacheId);
                    }
                    if (failed.get(cacheId) != null) {
                        Thread.sleep(1000);
                    }
                    Thread.sleep(100 + new Random().nextLong(400));
                } catch (InterruptedException ex) {
                    logger.error("Interrupted while waiting to refresh RDF cache: {}", ex.getMessage());
                }
                try {
                    updateRdfModel(queryRef);
                    failed.remove(cacheId);
                } catch (Exception ex) {
                    logger.error("Failed to update RDF cache for {}: {}", cacheId, ex.getMessage());
                    if (cachedRdfModels.getIfPresent(cacheId) == null) {
                        failed.merge(cacheId, 1, Integer::sum);
                    }
                    lastRefresh.put(cacheId, System.currentTimeMillis());
                } finally {
                    refreshStart.remove(cacheId);
                }
            });
        }
        if (isCached) {
            return cachedRdfModels.getIfPresent(cacheId);
        } else {
            return null;
        }
    }

    /**
     * Clears the cached response for a specific query reference and sets a delay before the next refresh can occur.
     *
     * @param queryRef   The query reference for which to clear the cache.
     * @param waitMillis The amount of time in milliseconds to wait before allowing the cache to be refreshed again.
     */
    public static void clearCache(QueryRef queryRef, long waitMillis) {
        if (waitMillis < 0) {
            throw new IllegalArgumentException("waitMillis must be non-negative");
        }
        cachedResponses.invalidate(queryRef.getAsUrlString());
        runAfter.put(queryRef.getAsUrlString(), System.currentTimeMillis() + waitMillis);
    }

}

1	package com.knowledgepixels.nanodash;
2
3	import com.google.common.cache.Cache;
4	import com.google.common.cache.CacheBuilder;
5	import com.google.common.cache.RemovalCause;
6	import com.google.common.cache.RemovalNotification;
7	import org.apache.wicket.MetaDataKey;
8	import org.apache.wicket.request.cycle.RequestCycle;
9	import org.eclipse.rdf4j.model.Model;
10	import org.nanopub.extra.services.*;
11	import org.slf4j.Logger;
12	import org.slf4j.LoggerFactory;
13
14	import java.util.HashMap;
15	import java.util.HashSet;
16	import java.util.List;
17	import java.util.Map;
18	import java.util.Random;
19	import java.util.concurrent.ConcurrentHashMap;
20	import java.util.concurrent.ConcurrentMap;
21	import java.util.concurrent.TimeUnit;
22
23	/**
24	* A utility class for caching API responses and maps to reduce redundant API calls.
25	* This class is thread-safe and ensures that cached data is refreshed periodically.
26	*/
27	public class ApiCache {
28
29	private ApiCache() {
30	} // no instances allowed
31
32	private static final int MAX_CACHE_ENTRIES = 10_000;
33
34	// How stale a cached response may be before the next access triggers a
35	// background re-fetch. Must stay reasonably high: every render that finds a
36	// query older than this submits a refresh to the shared pool, which uses a
37	// CallerRunsPolicy — so too low a value turns page renders into a refresh
38	// storm that can run queries synchronously on the request thread.
39	private static final long REFRESH_AGE_THRESHOLD_MS = 60 * 1000;
40
41	// How long a cached response is still served immediately (while refreshing in
42	// the background) before it is treated as absent and the caller waits for a
43	// fresh fetch. Acts as the stale-data fallback during API outages.
44	private static final long MAX_CACHE_AGE_MS = 24 * 60 * 60 * 1000;
45
46	// Upper bound a synchronous caller waits for an in-flight refresh started by
47	// another thread when it has nothing cached yet. Without this wait the caller
48	// returns null, letting repositories memoise an empty snapshot (see
49	// retrieveResponseSync).
50	private static final long SYNC_WAIT_FOR_INFLIGHT_MS = 10 * 1000;
51
52	private static final Cache<String, ApiResponse> cachedResponses = CacheBuilder.newBuilder()	6✔
53	.maximumSize(MAX_CACHE_ENTRIES)	9✔
54	.expireAfterAccess(24, TimeUnit.HOURS)	6✔
55	.removalListener(ApiCache::cleanupMetadataOnRemoval)	3✔
56	.build();	6✔
57	private static final Cache<String, Model> cachedRdfModels = CacheBuilder.newBuilder()	6✔
58	.maximumSize(MAX_CACHE_ENTRIES)	9✔
59	.expireAfterAccess(24, TimeUnit.HOURS)	6✔
60	.removalListener(ApiCache::cleanupMetadataOnRemoval)	3✔
61	.build();	6✔
62	private transient static ConcurrentMap<String, Integer> failed = new ConcurrentHashMap<>();	12✔
63	private static final Cache<String, Map<String, String>> cachedMaps = CacheBuilder.newBuilder()	6✔
64	.maximumSize(MAX_CACHE_ENTRIES)	9✔
65	.expireAfterAccess(24, TimeUnit.HOURS)	6✔
66	.removalListener(ApiCache::cleanupMetadataOnRemoval)	3✔
67	.build();	6✔
68	private transient static ConcurrentMap<String, Long> lastRefresh = new ConcurrentHashMap<>();	12✔
69	private transient static ConcurrentMap<String, Long> refreshStart = new ConcurrentHashMap<>();	12✔
70	private transient static ConcurrentMap<String, Long> runAfter = new ConcurrentHashMap<>();	12✔
71	private static final Logger logger = LoggerFactory.getLogger(ApiCache.class);	9✔
72
73	// Guava fires removal notifications also when an entry is REPLACED (every routine
74	// refresh's put), and processes them lazily during later cache operations. Cleaning
75	// up on a replacement would wipe the metadata of a still-cached entry — in
76	// particular a missing lastRefresh timestamp used to make retrieveResponseSync
77	// throw an NPE on every call until the entry expired.
78	private static void cleanupMetadataOnRemoval(RemovalNotification<String, ?> n) {
79	if (n.getCause() == RemovalCause.REPLACED) return;	15✔
80	cleanupMetadata(n.getKey());	12✔
81	}	3✔
82
83	private static void cleanupMetadata(String cacheId) {
84	lastRefresh.remove(cacheId);	12✔
85	failed.remove(cacheId);	12✔
86	runAfter.remove(cacheId);	12✔
87	}	3✔
88
89	/**
90	* Checks if a cache refresh is currently running for the given cache ID.
91	*
92	* @param cacheId The unique identifier for the cache.
93	* @return True if a refresh is running, false otherwise.
94	*/
95	private static boolean isRunning(String cacheId) {
96	Long start = refreshStart.get(cacheId);	15✔
97	if (start == null) return false;	12✔
98	return System.currentTimeMillis() - start < 60 * 1000;	33✔
99	}
100
101	/**
102	* Checks if a cache refresh is currently running for the given QueryRef.
103	*
104	* @param queryRef The query reference
105	* @return True if a refresh is running, false otherwise.
106	*/
107	public static boolean isRunning(QueryRef queryRef) {
108	return isRunning(queryRef.getAsUrlString());	12✔
109	}
110
111	/**
112	* Request-scoped flag set by {@code NanodashPage} when the current request is a
113	* genuine browser reload (the browser sends {@code Cache-Control: max-age=0} or
114	* {@code no-cache}). When set, the first access to each query during the page
115	* render evicts that query's cache so it re-fetches fresh, while normal
116	* navigation, Ajax updates, and the auto-refresh redirect keep serving the
117	* cache. Public so the page layer can set it.
118	*/
119	public static final MetaDataKey<Boolean> FORCE_REFRESH_ON_RELOAD = new MetaDataKey<>() {};	21✔
120
121	// The query cache-ids already force-evicted during the current reload request,
122	// so each is evicted only once (the lazy-load that follows must not re-evict).
123	private static final MetaDataKey<HashSet<String>> RELOAD_FORCED_IDS = new MetaDataKey<>() {};	24✔
124
125	/**
126	* On a genuine browser reload, returns true the first time a given query is
127	* accessed this request (and records it), so callers evict its cache once.
128	* Returns false on non-reload requests, off the request thread, and for any
129	* query already handled this request — so it never triggers a refresh storm.
130	*/
131	private static boolean isForcedReload(String cacheId) {
132	RequestCycle rc = RequestCycle.get();	6✔
133	if (rc == null) return false;	6!
134	Boolean force = rc.getMetaData(FORCE_REFRESH_ON_RELOAD);	15✔
135	if (force == null \|\| !force) return false;	12!
136	HashSet<String> handled = rc.getMetaData(RELOAD_FORCED_IDS);	×
137	if (handled == null) {	×
138	handled = new HashSet<>();	×
139	rc.setMetaData(RELOAD_FORCED_IDS, handled);	×
140	}
141	return handled.add(cacheId);	×
142	}
143
144	/**
145	* Updates the cached API response for a specific query reference.
146	*
147	* @param queryRef The query reference
148	* @throws FailedApiCallException If the API call fails.
149	*/
150	private static void updateResponse(QueryRef queryRef, boolean forced) throws FailedApiCallException, APINotReachableException, NotEnoughAPIInstancesException {
151	ApiResponse response;
152	if (forced) {	6✔
153	response = QueryApiAccess.forcedGet(queryRef);	12✔
154	} else {
155	response = QueryApiAccess.get(queryRef);	9✔
156	}
157	String cacheId = queryRef.getAsUrlString();	9✔
158	logger.info("Updating cached API response for {}", cacheId);	12✔
159	cachedResponses.put(cacheId, response);	12✔
160	lastRefresh.put(cacheId, System.currentTimeMillis());	18✔
161	}	3✔
162
163	public static ApiResponse retrieveResponseSync(QueryRef queryRef, boolean forced) {
164	long timeNow = System.currentTimeMillis();	6✔
165	String cacheId = queryRef.getAsUrlString();	9✔
166	logger.debug("Retrieving cached API response synchronously for {}", cacheId);	12✔
167	boolean needsRefresh = true;	6✔
168	if (cachedResponses.getIfPresent(cacheId) != null) {	12✔
169	// lastRefresh can be missing for a cached entry (racing invalidation or
170	// refresh); treat that as stale rather than NPEing on the unboxing.
171	Long lastRefreshTime = lastRefresh.get(cacheId);	15✔
172	needsRefresh = lastRefreshTime == null \|\| timeNow - lastRefreshTime > REFRESH_AGE_THRESHOLD_MS;	39!
173	}
174	Integer failedCount = failed.get(cacheId);	15✔
175	if (failedCount != null && failedCount > 2) {	18!
176	failed.remove(cacheId);	12✔
177	throw new RuntimeException("Query failed: " + cacheId);	18✔
178	}
179	if ((needsRefresh \|\| forced) && !isRunning(cacheId)) {	21!
180	logger.info("Refreshing cache for {}", cacheId);	12✔
181	refreshStart.put(cacheId, timeNow);	18✔
182	try {
183	Long after = runAfter.get(cacheId);	15✔
184	if (after != null) {	6!
185	while (System.currentTimeMillis() < after) {	×
186	Thread.sleep(100);	×
187	}
188	runAfter.remove(cacheId);	×
189	}
190	if (failed.get(cacheId) != null) {	12!
191	// 1 second pause between failed attempts;
192	Thread.sleep(1000);	×
193	}
194	Thread.sleep(100 + new Random().nextLong(400));	24✔
195	} catch (InterruptedException ex) {	×
196	logger.error("Interrupted while waiting to refresh cache: {}", ex.getMessage());	×
197	}	3✔
198	try {
199	ApiCache.updateResponse(queryRef, forced);	9✔
200	failed.remove(cacheId);	12✔
201	} catch (Exception ex) {	3✔
202	logger.error("Failed to update cache for {}: {}", cacheId, ex.getMessage());	18✔
203	// Keep stale cached data if available, only invalidate if nothing was cached
204	if (cachedResponses.getIfPresent(cacheId) == null) {	12!
205	failed.merge(cacheId, 1, Integer::sum);	21✔
206	}
207	lastRefresh.put(cacheId, System.currentTimeMillis());	18✔
208	} finally {
209	refreshStart.remove(cacheId);	12✔
210	}	3✔
211	} else if (cachedResponses.getIfPresent(cacheId) == null && isRunning(cacheId)) {	12!
212	// Another thread is doing the first fetch of this query and we have
213	// nothing cached yet. Wait for it rather than returning null: a null
214	// here lets a caller (e.g. SpaceRepository) memoise an EMPTY snapshot,
215	// which then poisons MaintainedResourceRepository.build() and breaks
216	// the home page until the next refresh. This adds no new work; it only
217	// waits on the refresh already in flight.
218	try {
219	long deadline = timeNow + SYNC_WAIT_FOR_INFLIGHT_MS;	×
220	while (isRunning(cacheId)	×
221	&& cachedResponses.getIfPresent(cacheId) == null	×
222	&& System.currentTimeMillis() < deadline) {	×
223	Thread.sleep(50);	×
224	}
225	} catch (InterruptedException ex) {	×
226	Thread.currentThread().interrupt();	×
227	}	×
228	}
229	return cachedResponses.getIfPresent(cacheId);	15✔
230	}
231
232	/**
233	* Retrieves a cached API response for a specific QueryRef.
234	*
235	* @param queryRef The QueryRef object containing the query name and parameters.
236	* @return The cached API response, or null if not cached.
237	*/
238	public static ApiResponse retrieveResponseAsync(QueryRef queryRef) {
239	long timeNow = System.currentTimeMillis();	6✔
240	String cacheId = queryRef.getAsUrlString();	9✔
241	logger.debug("Retrieving cached API response asynchronously for {}", cacheId);	12✔
242	if (isForcedReload(cacheId)) {	9!
243	cachedResponses.invalidate(cacheId);	×
244	lastRefresh.remove(cacheId);	×
245	}
246	boolean isCached = false;	6✔
247	boolean needsRefresh = true;	6✔
248	if (cachedResponses.getIfPresent(cacheId) != null) {	12✔
249	Long lastRefreshTime = lastRefresh.get(cacheId);	15✔
250	isCached = lastRefreshTime != null && timeNow - lastRefreshTime < MAX_CACHE_AGE_MS;	36!
251	needsRefresh = lastRefreshTime == null \|\| timeNow - lastRefreshTime > REFRESH_AGE_THRESHOLD_MS;	33!
252	}
253	Integer failedCount = failed.get(cacheId);	15✔
254	if (failedCount != null && failedCount > 2) {	6!
255	failed.remove(cacheId);	×
256	throw new RuntimeException("Query failed: " + cacheId);	×
257	}
258	if (needsRefresh && !isRunning(cacheId)) {	15!
259	NanodashThreadPool.submit(() -> {	15✔
260	refreshStart.put(cacheId, System.currentTimeMillis());	18✔
261	try {
262	Long after = runAfter.get(cacheId);	15✔
263	if (after != null) {	6!
264	while (System.currentTimeMillis() < after) {	×
265	Thread.sleep(100);	×
266	}
267	runAfter.remove(cacheId);	×
268	}
269	if (failed.get(cacheId) != null) {	12!
270	// 1 second pause between failed attempts;
271	Thread.sleep(1000);	×
272	}
273	Thread.sleep(100 + new Random().nextLong(400));	24✔
274	} catch (InterruptedException ex) {	×
275	logger.error("Interrupted while waiting to refresh cache: {}", ex.getMessage());	×
276	}	3✔
277	try {
278	ApiCache.updateResponse(queryRef, false);	9✔
279	failed.remove(cacheId);	12✔
280	} catch (Exception ex) {	×
281	logger.error("Failed to update cache for {}: {}", cacheId, ex.getMessage());	×
282	if (cachedResponses.getIfPresent(cacheId) == null) {	×
283	failed.merge(cacheId, 1, Integer::sum);	×
284	}
285	lastRefresh.put(cacheId, System.currentTimeMillis());	×
286	} finally {
287	refreshStart.remove(cacheId);	12✔
288	}
289	});	3✔
290	}
291	if (isCached) {	6✔
292	return cachedResponses.getIfPresent(cacheId);	15✔
293	} else {
294	return null;	6✔
295	}
296	}
297
298	/**
299	* Updates the cached map for a specific query reference.
300	*
301	* @param queryRef The query reference
302	* @throws FailedApiCallException If the API call fails.
303	*/
304	private static void updateMap(QueryRef queryRef) throws FailedApiCallException, APINotReachableException, NotEnoughAPIInstancesException {
305	Map<String, String> map = new HashMap<>();	×
306	List<ApiResponseEntry> respList = QueryApiAccess.get(queryRef).getData();	×
307	while (respList != null && !respList.isEmpty()) {	×
308	ApiResponseEntry resultEntry = respList.removeFirst();	×
309	map.put(resultEntry.get("key"), resultEntry.get("value"));	×
310	}	×
311	String cacheId = queryRef.getAsUrlString();	×
312	cachedMaps.put(cacheId, map);	×
313	lastRefresh.put(cacheId, System.currentTimeMillis());	×
314	}	×
315
316	/**
317	* Retrieves a cached map for a specific query reference.
318	* If the cache is stale, it triggers a background refresh.
319	*
320	* @param queryRef The query reference
321	* @return The cached map, or null if not cached.
322	*/
323	public static Map<String, String> retrieveMap(QueryRef queryRef) {
324	long timeNow = System.currentTimeMillis();	×
325	String cacheId = queryRef.getAsUrlString();	×
326	if (isForcedReload(cacheId)) {	×
327	cachedMaps.invalidate(cacheId);	×
328	lastRefresh.remove(cacheId);	×
329	}
330	boolean isCached = false;	×
331	boolean needsRefresh = true;	×
332	if (cachedMaps.getIfPresent(cacheId) != null) {	×
333	Long lastRefreshTime = lastRefresh.get(cacheId);	×
334	isCached = lastRefreshTime != null && timeNow - lastRefreshTime < MAX_CACHE_AGE_MS;	×
335	needsRefresh = lastRefreshTime == null \|\| timeNow - lastRefreshTime > REFRESH_AGE_THRESHOLD_MS;	×
336	}
337	if (needsRefresh && !isRunning(cacheId)) {	×
338	NanodashThreadPool.submit(() -> {	×
339	refreshStart.put(cacheId, System.currentTimeMillis());	×
340	try {
341	Long after = runAfter.get(cacheId);	×
342	if (after != null) {	×
343	while (System.currentTimeMillis() < after) {	×
344	Thread.sleep(100);	×
345	}
346	runAfter.remove(cacheId);	×
347	}
348	Thread.sleep(100 + new Random().nextLong(400));	×
349	} catch (InterruptedException ex) {	×
350	logger.error("Interrupted while waiting to refresh cache: {}", ex.getMessage());	×
351	}	×
352	try {
353	ApiCache.updateMap(queryRef);	×
354	} catch (Exception ex) {	×
355	logger.error("Failed to update cache for {}: {}", cacheId, ex.getMessage());	×
356	cachedMaps.invalidate(cacheId);	×
357	lastRefresh.put(cacheId, System.currentTimeMillis());	×
358	} finally {
359	refreshStart.remove(cacheId);	×
360	}
361	});	×
362	}
363	if (isCached) {	×
364	return cachedMaps.getIfPresent(cacheId);	×
365	} else {
366	return null;	×
367	}
368	}
369
370	private static void updateRdfModel(QueryRef queryRef) throws FailedApiCallException, APINotReachableException, NotEnoughAPIInstancesException {
371	final Model[] modelRef = new Model[1];	×
372	QueryAccess qa = new QueryAccess() {	×
373	@Override
374	protected void processHeader(String[] line) {}	×
375	@Override
376	protected void processLine(String[] line) {}	×
377	@Override
378	protected void processRdfContent(Model model) {
379	modelRef[0] = model;	×
380	}	×
381	};
382	qa.call(queryRef);	×
383	if (modelRef[0] == null) {	×
384	throw new FailedApiCallException(new Exception("No RDF content in response for query: " + queryRef.getQueryId()));	×
385	}
386	String cacheId = queryRef.getAsUrlString();	×
387	logger.info("Updating cached RDF model for {}", cacheId);	×
388	cachedRdfModels.put(cacheId, modelRef[0]);	×
389	lastRefresh.put(cacheId, System.currentTimeMillis());	×
390	}	×
391
392	/**
393	* Retrieves a cached RDF model for a CONSTRUCT query, triggering a background fetch if needed.
394	*
395	* @param queryRef The QueryRef for the CONSTRUCT query.
396	* @return The cached RDF Model, or null if not yet available.
397	*/
398	public static Model retrieveRdfModelAsync(QueryRef queryRef) {
399	long timeNow = System.currentTimeMillis();	×
400	String cacheId = queryRef.getAsUrlString();	×
401	logger.debug("Retrieving cached RDF model asynchronously for {}", cacheId);	×
402	if (isForcedReload(cacheId)) {	×
403	cachedRdfModels.invalidate(cacheId);	×
404	lastRefresh.remove(cacheId);	×
405	}
406	boolean isCached = false;	×
407	boolean needsRefresh = true;	×
408	if (cachedRdfModels.getIfPresent(cacheId) != null) {	×
409	Long lastRefreshTime = lastRefresh.get(cacheId);	×
410	isCached = lastRefreshTime != null && timeNow - lastRefreshTime < MAX_CACHE_AGE_MS;	×
411	needsRefresh = lastRefreshTime == null \|\| timeNow - lastRefreshTime > REFRESH_AGE_THRESHOLD_MS;	×
412	}
413	Integer failedCount = failed.get(cacheId);	×
414	if (failedCount != null && failedCount > 2) {	×
415	failed.remove(cacheId);	×
416	throw new RuntimeException("Query failed: " + cacheId);	×
417	}
418	if (needsRefresh && !isRunning(cacheId)) {	×
419	NanodashThreadPool.submit(() -> {	×
420	refreshStart.put(cacheId, System.currentTimeMillis());	×
421	try {
422	Long after = runAfter.get(cacheId);	×
423	if (after != null) {	×
424	while (System.currentTimeMillis() < after) {	×
425	Thread.sleep(100);	×
426	}
427	runAfter.remove(cacheId);	×
428	}
429	if (failed.get(cacheId) != null) {	×
430	Thread.sleep(1000);	×
431	}
432	Thread.sleep(100 + new Random().nextLong(400));	×
433	} catch (InterruptedException ex) {	×
434	logger.error("Interrupted while waiting to refresh RDF cache: {}", ex.getMessage());	×
435	}	×
436	try {
437	updateRdfModel(queryRef);	×
438	failed.remove(cacheId);	×
439	} catch (Exception ex) {	×
440	logger.error("Failed to update RDF cache for {}: {}", cacheId, ex.getMessage());	×
441	if (cachedRdfModels.getIfPresent(cacheId) == null) {	×
442	failed.merge(cacheId, 1, Integer::sum);	×
443	}
444	lastRefresh.put(cacheId, System.currentTimeMillis());	×
445	} finally {
446	refreshStart.remove(cacheId);	×
447	}
448	});	×
449	}
450	if (isCached) {	×
451	return cachedRdfModels.getIfPresent(cacheId);	×
452	} else {
453	return null;	×
454	}
455	}
456
457	/**
458	* Clears the cached response for a specific query reference and sets a delay before the next refresh can occur.
459	*
460	* @param queryRef The query reference for which to clear the cache.
461	* @param waitMillis The amount of time in milliseconds to wait before allowing the cache to be refreshed again.
462	*/
463	public static void clearCache(QueryRef queryRef, long waitMillis) {
464	if (waitMillis < 0) {	12✔
465	throw new IllegalArgumentException("waitMillis must be non-negative");	15✔
466	}
467	cachedResponses.invalidate(queryRef.getAsUrlString());	12✔
468	runAfter.put(queryRef.getAsUrlString(), System.currentTimeMillis() + waitMillis);	27✔
469	}	3✔
470
471	}

knowledgepixels / nanodash / 29227150127

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous