28022637110

Committed 23 Jun 2026 11:23AM UTC coverage: 26.541% (-0.008%) from 26.549%

Build # 28022637110

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #495 from knowledgepixels/fix/home-page-cold-cache-race

fix: prevent cold-cache race that breaks the home page

Coverage Stats

1555 of 6905 branches covered (22.52%)

Branch coverage included in aggregate %.

3423 of 11851 relevant lines covered (28.88%)

4.25 hits per line

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

39.27

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 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(n -> cleanupMetadata(n.getKey().toString()))
        .build();
    private static final Cache<String, Model> cachedRdfModels = CacheBuilder.newBuilder()
        .maximumSize(MAX_CACHE_ENTRIES)
        .expireAfterAccess(24, TimeUnit.HOURS)
        .removalListener(n -> cleanupMetadata(n.getKey().toString()))
        .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(n -> cleanupMetadata(n.getKey().toString()))
        .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);

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

knowledgepixels / nanodash / 28022637110

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