Skip to content

fix: release post-publish refreshes on ingest confirmation instead of a blind delay #645

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tkuhn merged 1 commit into from
Aug 25, 2026
Merged

fix: release post-publish refreshes on ingest confirmation instead of a blind delay #645

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
147 changes: 117 additions & 30 deletions src/main/java/com/knowledgepixels/nanodash/ApiCache.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,29 @@ private ApiCache() {
// attempt has completed, successfully or not.
private static final Set<String> 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<String, String> 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<String, Boolean> ingestConfirmResults = CacheBuilder.newBuilder()
.maximumSize(1000)
.expireAfterWrite(60, TimeUnit.SECONDS)
.build();
private transient static ConcurrentMap<String, Object> ingestConfirmLocks = new ConcurrentHashMap<>();

private static final Logger logger = LoggerFactory.getLogger(ApiCache.class);

// Guava fires removal notifications also when an entry is REPLACED (every routine
Expand All @@ -96,6 +119,7 @@ private static void cleanupMetadata(String cacheId) {
failed.remove(cacheId);
runAfter.remove(cacheId);
forcedRefresh.remove(cacheId);
awaitIngest.remove(cacheId);
}

/**
Expand Down Expand Up @@ -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.
*
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -446,13 +528,7 @@ public static Map<String, String> 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());
Expand Down Expand Up @@ -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);
}
Expand Down Expand Up @@ -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);
}

}
19 changes: 19 additions & 0 deletions src/main/java/com/knowledgepixels/nanodash/QueryApiAccess.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
*
Expand Down
2 changes: 1 addition & 1 deletion src/main/java/com/knowledgepixels/nanodash/WicketApplication.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -362,7 +362,7 @@ private void registerListeners() {
}
} else {
QueryRef queryRef = QueryRef.parseString(target);
ApiCache.clearCache(queryRef, waitMs);
ApiCache.clearCache(queryRef, waitMs, nanopub.getUri().stringValue());
}
});
}
Expand Down
73 changes: 73 additions & 0 deletions src/test/java/com/knowledgepixels/nanodash/ApiCacheTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
}
Expand Down Expand Up @@ -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> 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.<String, String>getMap("awaitIngest").containsKey(MOCK_CACHE_ID), "the pending confirmation should be consumed");
assertFalse(this.<String, Long>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> 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.<String, Boolean>getMap("ingestConfirmResults").put(npId, false);
ApiCache.clearCache(mockQueryRef, 0L, npId);

try (MockedStatic<QueryApiAccess> 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 {
Expand Down