From dc893944145f7b42618bde5fa7038a2cd2a5e21f Mon Sep 17 00:00:00 2001 From: Tobias Kuhn Date: Tue, 25 Aug 2026 14:25:21 +0200 Subject: [PATCH] fix: release post-publish refreshes on ingest confirmation instead of a blind delay The refresh after publishing slept a fixed 5 seconds, fetched once, and gave up: if the nanopub had not been ingested by then, the one fetch returned the old data and the view settled on stale content until a manual reload (issue #629). The wait is now a measurement: clearCache can be handed the just-published nanopub's id, and the background refresh polls the new check-nanopub-loaded query (a single indexed lookup on the meta repo) until the query services report the nanopub as loaded, plus a small margin for trailing repos and instances. The polling is hard-bounded (20s cap, 1s interval), shared between the several views refreshing after the same publish, and never runs on a request thread. If the probe fails or times out, the blind fallback delay applies unchanged, so a broken probe cannot make publishing worse than before. Fixes #629 Co-Authored-By: Claude Fable 5 --- .../knowledgepixels/nanodash/ApiCache.java | 147 ++++++++++++++---- .../nanodash/QueryApiAccess.java | 19 +++ .../nanodash/WicketApplication.java | 2 +- .../nanodash/ApiCacheTest.java | 73 +++++++++ 4 files changed, 210 insertions(+), 31 deletions(-) diff --git a/src/main/java/com/knowledgepixels/nanodash/ApiCache.java b/src/main/java/com/knowledgepixels/nanodash/ApiCache.java index dae59986..c95bd1ae 100644 --- a/src/main/java/com/knowledgepixels/nanodash/ApiCache.java +++ b/src/main/java/com/knowledgepixels/nanodash/ApiCache.java @@ -79,6 +79,29 @@ private ApiCache() { // attempt has completed, successfully or not. private static final Set forcedRefresh = ConcurrentHashMap.newKeySet(); + // How long we keep polling for a just-published nanopub to show up at the query + // services before giving up and refreshing anyway (issue #629). A hard bound: the + // probe is a single indexed lookup, but an unbounded retry loop from many publishing + // sessions is the load shape that has wedged the query API before. + private static final long INGEST_CONFIRM_MAX_WAIT_MS = 20 * 1000; + private static final long INGEST_CONFIRM_POLL_INTERVAL_MS = 1000; + // Margin after a positive probe: the confirming instance has the nanopub, but its + // other repos and the other instances may trail slightly behind. + private static final long INGEST_CONFIRM_MARGIN_MS = 1000; + + // Cache ids whose next refresh should wait for the given nanopub to be ingested + // rather than (only) sit out the blind runAfter delay; set by clearCache after a + // publish, consumed by waitOutIngestDelay in the background refresh. + private transient static ConcurrentMap awaitIngest = new ConcurrentHashMap<>(); + // Shared probe results, so several views refreshing after the same publish cost one + // polling loop, not one each. False (timed out or probe failed) is cached too, to + // keep late arrivals from re-running a full polling round that already gave up. + private static final Cache ingestConfirmResults = CacheBuilder.newBuilder() + .maximumSize(1000) + .expireAfterWrite(60, TimeUnit.SECONDS) + .build(); + private transient static ConcurrentMap ingestConfirmLocks = new ConcurrentHashMap<>(); + private static final Logger logger = LoggerFactory.getLogger(ApiCache.class); // Guava fires removal notifications also when an entry is REPLACED (every routine @@ -96,6 +119,7 @@ private static void cleanupMetadata(String cacheId) { failed.remove(cacheId); runAfter.remove(cacheId); forcedRefresh.remove(cacheId); + awaitIngest.remove(cacheId); } /** @@ -191,6 +215,74 @@ private static boolean isForcedReload(String cacheId) { return handled.add(cacheId); } + /** + * Waits out the post-publish ingest delay for a cache entry, if one is pending, + * before its refresh is allowed to run. With a nanopub to wait for (see + * {@link #clearCache(QueryRef, long, String)}), the wait is a measurement: poll + * until the query services report the nanopub as loaded, plus a small margin. If + * there is none, or the probe fails or times out, this falls back to the blind + * runAfter delay, so a broken probe never makes publishing worse than before. + * Runs on background threads only; request threads are diverted beforehand. + * + * @param cacheId the cache id (the query's URL string) + */ + private static void waitOutIngestDelay(String cacheId) throws InterruptedException { + String npId = awaitIngest.remove(cacheId); + if (npId != null && awaitNanopubLoaded(npId)) { + Thread.sleep(INGEST_CONFIRM_MARGIN_MS); + runAfter.remove(cacheId); + return; + } + Long after = runAfter.get(cacheId); + if (after != null) { + while (System.currentTimeMillis() < after) { + Thread.sleep(100); + } + runAfter.remove(cacheId); + } + } + + /** + * Polls the query services until they report the given nanopub as loaded, bounded by + * {@link #INGEST_CONFIRM_MAX_WAIT_MS}. Concurrent callers for the same nanopub (the + * several views refreshing after one publish) share a single polling loop: the first + * caller polls, the others wait on its result. + * + * @param npId the nanopub id to wait for + * @return true if the nanopub was confirmed as loaded, false if the probe timed out + * or failed (callers then fall back to the blind delay) + */ + private static boolean awaitNanopubLoaded(String npId) throws InterruptedException { + Boolean known = ingestConfirmResults.getIfPresent(npId); + if (known != null) return known; + Object lock = ingestConfirmLocks.computeIfAbsent(npId, k -> new Object()); + synchronized (lock) { + try { + known = ingestConfirmResults.getIfPresent(npId); + if (known != null) return known; + long deadline = System.currentTimeMillis() + INGEST_CONFIRM_MAX_WAIT_MS; + boolean loaded = false; + while (true) { + try { + loaded = QueryApiAccess.isNanopubLoaded(npId); + } catch (Exception ex) { + logger.warn("Nanopub load probe failed for {}: {}", npId, ex.getMessage()); + break; + } + if (loaded || System.currentTimeMillis() + INGEST_CONFIRM_POLL_INTERVAL_MS > deadline) break; + Thread.sleep(INGEST_CONFIRM_POLL_INTERVAL_MS); + } + if (!loaded) { + logger.info("Nanopub {} not confirmed as loaded, falling back to blind delay", npId); + } + ingestConfirmResults.put(npId, loaded); + return loaded; + } finally { + ingestConfirmLocks.remove(npId, lock); + } + } + } + /** * Updates the cached API response for a specific query reference. * @@ -240,7 +332,8 @@ public static ApiResponse retrieveResponseSync(QueryRef queryRef, boolean forced // and leaves the refresh to a thread that can afford to wait. boolean onRequestThread = RequestCycle.get() != null; Long after = runAfter.get(cacheId); - boolean waitingForIngest = after != null && System.currentTimeMillis() < after; + boolean waitingForIngest = (after != null && System.currentTimeMillis() < after) + || awaitIngest.containsKey(cacheId); if (onRequestThread && waitingForIngest) { logger.debug("Not waiting out the ingest delay for {} on a request thread", cacheId); // Hand the refresh to the background, where waiting out the delay costs nobody @@ -269,12 +362,7 @@ public static ApiResponse retrieveResponseSync(QueryRef queryRef, boolean forced logger.info("Refreshing cache for {}", cacheId); refreshStart.put(cacheId, timeNow); try { - if (waitingForIngest) { - while (System.currentTimeMillis() < after) { - Thread.sleep(100); - } - } - if (after != null) runAfter.remove(cacheId); + waitOutIngestDelay(cacheId); if (!onRequestThread) { if (failed.get(cacheId) != null) { // 1 second pause between failed attempts; @@ -359,13 +447,7 @@ public static ApiResponse retrieveResponseAsync(QueryRef queryRef) { 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); - } + waitOutIngestDelay(cacheId); if (failed.get(cacheId) != null) { // 1 second pause between failed attempts; Thread.sleep(1000); @@ -446,13 +528,7 @@ public static Map retrieveMap(QueryRef queryRef) { 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); - } + waitOutIngestDelay(cacheId); Thread.sleep(100 + new Random().nextLong(400)); } catch (InterruptedException ex) { logger.error("Interrupted while waiting to refresh cache: {}", ex.getMessage()); @@ -532,13 +608,7 @@ public static Model retrieveRdfModelAsync(QueryRef queryRef) { 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); - } + waitOutIngestDelay(cacheId); if (failed.get(cacheId) != null) { Thread.sleep(1000); } @@ -711,11 +781,28 @@ static void backfillEntryStore(Snapshot snapshot) { * @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) { + clearCache(queryRef, waitMillis, null); + } + + /** + * Like {@link #clearCache(QueryRef, long)}, but for the refresh after a publish: the + * refresh is released as soon as the query services confirm the given nanopub as + * loaded (plus a small margin), instead of after the blind delay (issue #629). The + * delay stays in place as the fallback for when the confirmation probe fails, and the + * confirmation wait itself is bounded by {@link #INGEST_CONFIRM_MAX_WAIT_MS}. + * + * @param queryRef The query reference for which to clear the cache. + * @param waitMillis The fallback delay in milliseconds, used if the nanopub's arrival cannot be confirmed. + * @param nanopubId The id of the just-published nanopub to wait for, or null for the plain delay. + */ + public static void clearCache(QueryRef queryRef, long waitMillis, String nanopubId) { if (waitMillis < 0) { throw new IllegalArgumentException("waitMillis must be non-negative"); } - forcedRefresh.add(queryRef.getAsUrlString()); - runAfter.put(queryRef.getAsUrlString(), System.currentTimeMillis() + waitMillis); + String cacheId = queryRef.getAsUrlString(); + forcedRefresh.add(cacheId); + runAfter.put(cacheId, System.currentTimeMillis() + waitMillis); + if (nanopubId != null) awaitIngest.put(cacheId, nanopubId); } } diff --git a/src/main/java/com/knowledgepixels/nanodash/QueryApiAccess.java b/src/main/java/com/knowledgepixels/nanodash/QueryApiAccess.java index ba13cae9..c51b161c 100644 --- a/src/main/java/com/knowledgepixels/nanodash/QueryApiAccess.java +++ b/src/main/java/com/knowledgepixels/nanodash/QueryApiAccess.java @@ -30,6 +30,9 @@ private QueryApiAccess() { public static final String GET_MOST_USED_TEMPLATES_LAST30D = "RAvL7pe2ppsfq4mVWTdJjssYGsjrmliNd_sZO2ytLvg1Y/get-most-used-templates-last30d"; public static final String GET_LATEST_NANOPUBS_BY_TYPE = "RANn4Mu8r8bqJA9KJMGXTQAEGAEvtNKGFsuhRIC6BRIOo/get-latest-nanopubs-by-type"; public static final String GET_LATEST_VERSION_OF_NP = "RAiRsB2YywxjsBMkVRTREJBooXhf2ZOHoUs5lxciEl37I/get-latest-version-of-np"; + // Minimal single-lookup probe for whether a given nanopub has been loaded by the + // answering Nanopub Query instance; used to time post-publish refreshes (issue #629). + public static final String CHECK_NANOPUB_LOADED = "RAxqXyhP1fnjvDdX-K0z9TgnwoXf462FxV1wEAWRm_gos/check-nanopub-loaded"; public static final String GET_ALL_USER_INTROS = "RAjHh6P11QFUaoPiMRBavdAnTq4YMJW4PB85oVFSBfYjU/get-all-user-intros"; public static final String GET_ALL_USER_PROFILE_PICS = "RAtcodMPmTrmBvdOqwYIrNNFDO74f8B_xo0qsOcKlCwTA/get-all-user-profile-pics"; // Profile pictures of spaces and maintained resources (issue #632), declared as @@ -400,6 +403,22 @@ public static String getLatestVersionId(String nanopubId) { return cached != null ? cached.getRight() : nanopubId; } + /** + * Checks whether the given nanopublication has been loaded by the query services, + * with a single cheap indexed lookup. A negative answer only means the instance that + * happened to answer does not have the nanopub yet. + * + * @param nanopubId The ID of the nanopublication. + * @return True if the answering query service instance has the nanopub. + * @throws org.nanopub.extra.services.FailedApiCallException If the API call fails. + * @throws org.nanopub.extra.services.APINotReachableException If the API is not reachable. + * @throws org.nanopub.extra.services.NotEnoughAPIInstancesException If there are not enough API instances. + */ + public static boolean isNanopubLoaded(String nanopubId) throws FailedApiCallException, APINotReachableException, NotEnoughAPIInstancesException { + ApiResponse r = get(new QueryRef(CHECK_NANOPUB_LOADED, "np", nanopubId)); + return r != null && !r.getData().isEmpty(); + } + /** * Extracts the query ID from a given query IRI. * diff --git a/src/main/java/com/knowledgepixels/nanodash/WicketApplication.java b/src/main/java/com/knowledgepixels/nanodash/WicketApplication.java index f816ecc6..ad32712a 100644 --- a/src/main/java/com/knowledgepixels/nanodash/WicketApplication.java +++ b/src/main/java/com/knowledgepixels/nanodash/WicketApplication.java @@ -362,7 +362,7 @@ private void registerListeners() { } } else { QueryRef queryRef = QueryRef.parseString(target); - ApiCache.clearCache(queryRef, waitMs); + ApiCache.clearCache(queryRef, waitMs, nanopub.getUri().stringValue()); } }); } diff --git a/src/test/java/com/knowledgepixels/nanodash/ApiCacheTest.java b/src/test/java/com/knowledgepixels/nanodash/ApiCacheTest.java index d24e52eb..76bb54b9 100644 --- a/src/test/java/com/knowledgepixels/nanodash/ApiCacheTest.java +++ b/src/test/java/com/knowledgepixels/nanodash/ApiCacheTest.java @@ -39,6 +39,9 @@ void setUp() throws Exception { resetMap("refreshStart"); resetMap("runAfter"); resetMap("forcedRefresh"); + resetMap("awaitIngest"); + resetMap("ingestConfirmResults"); + resetMap("ingestConfirmLocks"); lenient().when(mockQueryRef.getAsUrlString()).thenReturn(MOCK_CACHE_ID); } @@ -334,6 +337,76 @@ void retrieveResponseSync_dropsMarkingWhenRefreshFails() throws Exception { } } + @Test + @DisplayName("a publish-marked refresh should be released by ingest confirmation instead of the blind delay") + void publishMarkedRefreshReleasedByIngestConfirmation() throws Exception { + ApiResponse stale = mock(ApiResponse.class); + ApiResponse fresh = mock(ApiResponse.class); + putCachedResponse(stale, 5000L); + String npId = "https://w3id.org/np/RAtest0000000000000000000000000000000000000x1"; + // A fallback delay far longer than the test may take: with a positive probe, the + // refresh must not wait it out. + ApiCache.clearCache(mockQueryRef, 60000L, npId); + + try (MockedStatic queryApiAccess = mockStatic(QueryApiAccess.class)) { + queryApiAccess.when(() -> QueryApiAccess.isNanopubLoaded(npId)).thenReturn(true); + queryApiAccess.when(() -> QueryApiAccess.get(mockQueryRef)).thenReturn(fresh); + + long start = System.currentTimeMillis(); + ApiResponse result = ApiCache.retrieveResponseSync(mockQueryRef, false); + long elapsed = System.currentTimeMillis() - start; + + assertSame(fresh, result); + assertTrue(elapsed < 10000, "confirmation should release the refresh long before the 60s fallback, but took " + elapsed + "ms"); + queryApiAccess.verify(() -> QueryApiAccess.isNanopubLoaded(npId)); + assertFalse(this.getMap("awaitIngest").containsKey(MOCK_CACHE_ID), "the pending confirmation should be consumed"); + assertFalse(this.getMap("runAfter").containsKey(MOCK_CACHE_ID), "the fallback delay should be dropped on confirmation"); + } + } + + @Test + @DisplayName("a failing ingest probe should fall back to the blind delay") + void failingIngestProbeFallsBackToBlindDelay() throws Exception { + ApiResponse stale = mock(ApiResponse.class); + ApiResponse fresh = mock(ApiResponse.class); + putCachedResponse(stale, 5000L); + String npId = "https://w3id.org/np/RAtest0000000000000000000000000000000000000x2"; + ApiCache.clearCache(mockQueryRef, 0L, npId); + + try (MockedStatic queryApiAccess = mockStatic(QueryApiAccess.class)) { + queryApiAccess.when(() -> QueryApiAccess.isNanopubLoaded(npId)).thenThrow(new FailedApiCallException(new Exception("probe broken"))); + queryApiAccess.when(() -> QueryApiAccess.get(mockQueryRef)).thenReturn(fresh); + + ApiResponse result = ApiCache.retrieveResponseSync(mockQueryRef, false); + + // A broken probe must never make publishing worse than before: the refresh + // still runs after the (here elapsed) fallback delay. + assertSame(fresh, result); + } + } + + @Test + @DisplayName("a shared negative probe result should short-circuit later waiters") + void sharedNegativeProbeResultShortCircuits() throws Exception { + ApiResponse stale = mock(ApiResponse.class); + ApiResponse fresh = mock(ApiResponse.class); + putCachedResponse(stale, 5000L); + String npId = "https://w3id.org/np/RAtest0000000000000000000000000000000000000x3"; + // Another view's refresh has already polled for this nanopub and given up. + this.getMap("ingestConfirmResults").put(npId, false); + ApiCache.clearCache(mockQueryRef, 0L, npId); + + try (MockedStatic queryApiAccess = mockStatic(QueryApiAccess.class)) { + queryApiAccess.when(() -> QueryApiAccess.get(mockQueryRef)).thenReturn(fresh); + + ApiResponse result = ApiCache.retrieveResponseSync(mockQueryRef, false); + + assertSame(fresh, result); + // The shared result answers instead of a second polling round. + queryApiAccess.verify(() -> QueryApiAccess.isNanopubLoaded(any()), never()); + } + } + @Test @DisplayName("clearCache should handle zero wait time") void clearCacheWithZeroWaitTime() throws Exception {