25035879209

Committed 28 Apr 2026 05:36AM UTC coverage: 56.362% (-0.4%) from 56.747%

Build # 25035879209

Build Type

Pull #91

github

Committed by

web-flow

Commit Message

Merge 1e84430c2 into a3228d481

Pull Request Pull Request #91: docs: drop SERVICE clause from worked-example query, names are now mirrored (#62)

Coverage Stats

425 of 842 branches covered (50.48%)

Branch coverage included in aggregate %.

1183 of 2011 relevant lines covered (58.83%)

9.0 hits per line

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

25.75

src/main/java/com/knowledgepixels/query/JellyNanopubLoader.java

package com.knowledgepixels.query;

import java.io.IOException;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpHead;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.util.EntityUtils;
import org.nanopub.NanopubUtils;
import org.nanopub.jelly.NanopubStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Loads nanopubs from the attached Nanopub Registry via a restartable Jelly stream.
 */
public class JellyNanopubLoader {
    static final String registryUrl;
    private static long lastCommittedCounter = -1;
    private static Long lastKnownSetupId = null;
    // Latest registry metadata fields, updated on each metadata fetch and forwarded to clients
    static volatile String lastCoverageTypes = null;
    static volatile String lastCoverageAgents = null;
    static volatile String lastTestInstance = null;
    static volatile String lastNanopubCount = null;
    private static final CloseableHttpClient metadataClient;
    private static final CloseableHttpClient jellyStreamClient;

    private static final int MAX_RETRIES_METADATA = 10;
    private static final int RETRY_DELAY_METADATA = 3000;
    private static final int RETRY_DELAY_JELLY = 5000;

    /**
     * Circuit-breaker state. Counts consecutive {@code loadUpdates} invocations in
     * which the catch block fired; resets to zero on the next successful batch.
     * Scheduled on the single-threaded executor in {@link MainVerticle}, so a plain
     * {@code int} is enough — no concurrency.
     *
     * <p>When the counter reaches {@link #BREAKER_THRESHOLD}, the next invocation
     * sleeps {@link #BREAKER_PAUSE_MS} before proceeding. Lets the saturated RDF4J
     * drain instead of being hammered every {@link #UPDATES_POLL_INTERVAL} ms.
     *
     * <p>Depends on {@link com.knowledgepixels.query.TripleStore}'s socket timeouts
     * (change 1 of the fix plan) to turn parked commits into propagating exceptions;
     * without them, {@code loadBatch} can park forever, the catch never fires, and
     * this counter stays at zero.
     */
    static volatile int consecutiveBatchFailures = 0;

    static final int BREAKER_THRESHOLD = 3;
    static final long BREAKER_PAUSE_MS = 30_000L;

    /**
     * Epoch-millis of the last successful {@code loadUpdates} invocation (whether it
     * actually loaded a batch or was a "caught up, nothing to do" tick). Read by
     * {@link MetricsCollector} to expose {@code registry.loader.last_successful_batch_age_seconds}
     * so an operator can tell at a glance from Grafana whether an instance is still
     * making progress. Updated on every non-exceptional return from loadUpdates,
     * including idle returns — an instance that is caught up and correctly polling
     * still counts as alive.
     */
    static volatile long lastSuccessfulBatchAtMs = 0L;

    /**
     * Heartbeat counter for loadUpdates invocations. A summary log line is emitted
     * every {@link #HEARTBEAT_INTERVAL_INVOCATIONS} invocations so a truncated or
     * sampled log still shows the loader's state evolving. At the default 2 s poll
     * interval, 30 invocations ≈ 1 line per minute.
     */
    private static long loadUpdatesInvocations = 0L;
    private static final long HEARTBEAT_INTERVAL_INVOCATIONS = 30;

    private static final Logger log = LoggerFactory.getLogger(JellyNanopubLoader.class);

    /**
     * Registry metadata returned by a HEAD request.
     */
    record RegistryMetadata(long loadCounter, Long setupId, String coverageTypes,
                            String coverageAgents, String testInstance, String nanopubCount,
                            String trustStateHash) {}

    /**
     * The interval in milliseconds at which the updates loader should poll for new nanopubs.
     */
    public static final int UPDATES_POLL_INTERVAL = 2000;

    enum LoadingType {
        INITIAL,
        UPDATE,
    }

    static {
        // Initialize registryUrl
        var url = Utils.getEnvString(
                "REGISTRY_FIXED_URL", "https://registry.knowledgepixels.com/"
        );
        if (!url.endsWith("/")) url += "/";
        registryUrl = url;

        metadataClient = HttpClientBuilder.create().setDefaultRequestConfig(Utils.getHttpRequestConfig()).build();
        jellyStreamClient = NanopubUtils.getHttpClient();
    }

    /**
     * Start or continue (after restart) the initial loading procedure. This simply loads all
     * nanopubs from the attached Registry.
     *
     * @param afterCounter which counter to start from (-1 for the beginning)
     */
    public static void loadInitial(long afterCounter) {
        RegistryMetadata metadata = fetchRegistryMetadata();
        updateForwardingMetadata(metadata);
        TrustStateLoader.maybeUpdate(metadata.trustStateHash());
        long targetCounter = metadata.loadCounter();
        log.info("Fetched Registry load counter: {}", targetCounter);
        // Store setupId on initial load
        if (metadata.setupId() != null && lastKnownSetupId == null) {
            lastKnownSetupId = metadata.setupId();
            StatusController.get().setRegistrySetupId(metadata.setupId());
        }
        lastCommittedCounter = afterCounter;
        while (lastCommittedCounter < targetCounter) {
            try {
                loadBatch(lastCommittedCounter, LoadingType.INITIAL);
                log.info("Initial load: loaded batch up to counter {}", lastCommittedCounter);
            } catch (Exception e) {
                log.info("Failed to load batch starting from counter {}", lastCommittedCounter);
                log.info("Failure reason: ", e);
                try {
                    Thread.sleep(RETRY_DELAY_JELLY);
                } catch (InterruptedException e2) {
                    throw new RuntimeException("Interrupted while waiting to retry loading batch.");
                }
            }
        }
        log.info("Initial load complete.");
    }

    /**
     * Check if the Registry has any new nanopubs. If it does, load them.
     * This method should be called periodically, and you should wait for it to finish before
     * calling it again.
     */
    public static void loadUpdates() {
        // Circuit breaker: after BREAKER_THRESHOLD consecutive failed batches, pause
        // before the next attempt so a saturated RDF4J can drain. Check happens before
        // any RDF4J-touching work so the sleep isn't itself under the broken regime.
        if (consecutiveBatchFailures >= BREAKER_THRESHOLD) {
            log.warn("Circuit breaker active after {} consecutive batch failures; pausing {} ms before next attempt",
                    consecutiveBatchFailures, BREAKER_PAUSE_MS);
            try {
                Thread.sleep(BREAKER_PAUSE_MS);
            } catch (InterruptedException e) {
                // Preserve interruption semantics so a graceful shutdown (e.g. via
                // MainVerticle's shutdown hook) isn't blocked by the pause.
                Thread.currentThread().interrupt();
                return;
            }
        }
        try {
            final var status = StatusController.get().getState();
            lastCommittedCounter = status.loadCounter;
            RegistryMetadata metadata = fetchRegistryMetadata();
            updateForwardingMetadata(metadata);
            TrustStateLoader.maybeUpdate(metadata.trustStateHash());
            long targetCounter = metadata.loadCounter();
            Long currentSetupId = metadata.setupId();

            // Detect reset via setupId change
            if (lastKnownSetupId != null && currentSetupId != null
                    && !lastKnownSetupId.equals(currentSetupId)) {
                log.warn("Registry reset detected: setupId {} -> {}", lastKnownSetupId, currentSetupId);
                performResync(currentSetupId);
                return;
            }
            // Detect reset via counter decrease (also covers first run after upgrade
            // where no setupId was persisted yet but the registry has already been reset)
            if (lastCommittedCounter > 0 && targetCounter >= 0
                    && targetCounter < lastCommittedCounter) {
                log.warn("Registry counter decreased {} -> {}, triggering resync",
                        lastCommittedCounter, targetCounter);
                performResync(currentSetupId);
                return;
            }

            // Update lastKnownSetupId on first successful poll
            if (currentSetupId != null && lastKnownSetupId == null) {
                if (lastCommittedCounter > 0) {
                    // Upgrade from a version without setupId tracking. The DB has data but
                    // we can't verify it matches the current registry. Force a resync.
                    log.warn("No stored setupId but DB has data (counter: {}). "
                            + "Forcing resync to ensure data consistency.", lastCommittedCounter);
                    performResync(currentSetupId);
                    return;
                }
                lastKnownSetupId = currentSetupId;
                StatusController.get().setRegistrySetupId(currentSetupId);
            }

            if (lastCommittedCounter >= targetCounter) {
                // Nothing to do. Keep state at READY (setReady is idempotent) and
                // skip the redundant setLoadingUpdates → setReady admin-repo write
                // that the old flow did on every idle poll. Also reset the breaker
                // counter — a successful "nothing to do" is still a successful tick
                // and should clear stale failure state from earlier transient errors.
                StatusController.get().setReady();
                consecutiveBatchFailures = 0;
                lastSuccessfulBatchAtMs = System.currentTimeMillis();
                maybeLogHeartbeat(targetCounter, true);
                return;
            }
            StatusController.get().setLoadingUpdates(status.loadCounter);
            loadBatch(lastCommittedCounter, LoadingType.UPDATE);
            // Batch completed without an exception — reset the breaker counter.
            consecutiveBatchFailures = 0;
            lastSuccessfulBatchAtMs = System.currentTimeMillis();
            maybeLogHeartbeat(targetCounter, false);
            log.info("Loaded {} update(s). Counter: {}, target was: {}",
                    lastCommittedCounter - status.loadCounter, lastCommittedCounter, targetCounter);
            if (lastCommittedCounter < targetCounter) {
                log.info("Warning: expected to load nanopubs up to (inclusive) counter " +
                        targetCounter + " based on the counter reported in Registry's headers, " +
                        "but loaded only up to {}.", lastCommittedCounter);
            }
        } catch (Exception e) {
            consecutiveBatchFailures++;
            log.warn("Failed to load updates. Current counter: {} (consecutive failures: {})",
                    lastCommittedCounter, consecutiveBatchFailures, e);
        } finally {
            try {
                StatusController.get().setReady();
            } catch (Exception e) {
                log.info("Update loader: failed to set status to READY.");
                log.info("Failure Reason: ", e);
            }
        }
    }

    /**
     * Emit a heartbeat summary log line roughly every
     * {@link #HEARTBEAT_INTERVAL_INVOCATIONS} invocations of {@link #loadUpdates}.
     * Lets an operator reconstruct loader progress from a sparse or sampled log
     * export, independent of Prometheus retention.
     */
    private static void maybeLogHeartbeat(long targetCounter, boolean idle) {
        loadUpdatesInvocations++;
        if (loadUpdatesInvocations % HEARTBEAT_INTERVAL_INVOCATIONS != 0) return;
        log.info("Loader heartbeat: counter={} target={} idle={} consecutiveBatchFailures={} breakerActive={}",
                lastCommittedCounter, targetCounter, idle, consecutiveBatchFailures,
                consecutiveBatchFailures >= BREAKER_THRESHOLD);
    }

    /**
     * Re-stream all nanopubs from the registry after a reset is detected.
     * Existing nanopubs are skipped by NanopubLoader's per-repo dedup.
     *
     * @param newSetupId the new setup ID from the registry, or null if unknown
     */
    private static void performResync(Long newSetupId) {
        log.warn("Starting resync with registry. New setupId: {}", newSetupId);
        StatusController.get().setResetting();
        lastKnownSetupId = newSetupId;
        if (newSetupId != null) {
            StatusController.get().setRegistrySetupId(newSetupId);
        }
        StatusController.get().setLoadingInitial(-1);
        loadInitial(-1);
        StatusController.get().setReady();
        log.warn("Resync complete. Counter: {}", lastCommittedCounter);
    }

    /**
     * Load a batch of nanopubs from the Jelly stream.
     * <p>
     * The method requests the list of all nanopubs from the Registry and reads it for as long
     * as it can. If the stream is interrupted, the method will throw an exception, and you
     * can resume loading from the last known counter.
     *
     * @param afterCounter the last known nanopub counter to have been committed in the DB
     * @param type         the type of loading operation (initial or update)
     */
    static void loadBatch(long afterCounter, LoadingType type) {
        CloseableHttpResponse response;
        try {
            var request = new HttpGet(makeStreamFetchUrl(afterCounter));
            response = jellyStreamClient.execute(request);
        } catch (IOException e) {
            throw new RuntimeException("Failed to fetch Jelly stream from the Registry (I/O error).", e);
        }

        int httpStatus = response.getStatusLine().getStatusCode();
        if (httpStatus < 200 || httpStatus >= 300) {
            EntityUtils.consumeQuietly(response.getEntity());
            throw new RuntimeException("Jelly stream HTTP status is not 2xx: " + httpStatus + ".");
        }

        try (
                var is = response.getEntity().getContent();
                var npStream = NanopubStream.fromByteStream(is).getAsNanopubs()
        ) {
            AtomicLong checkpointTime = new AtomicLong(System.currentTimeMillis());
            AtomicLong checkpointCounter = new AtomicLong(lastCommittedCounter);
            AtomicLong lastSavedCounter = new AtomicLong(lastCommittedCounter);
            AtomicLong loaded = new AtomicLong(0L);

            npStream.forEach(m -> {
                if (!m.isSuccess()) throw new RuntimeException("Failed to load " +
                        "nanopub from Jelly stream. Last known counter: " + lastCommittedCounter,
                        m.getException()
                );
                if (m.getCounter() < lastCommittedCounter) {
                    throw new RuntimeException("Received a nanopub with a counter lower than " +
                            "the last known counter. Last known counter: " + lastCommittedCounter +
                            ", received counter: " + m.getCounter());
                }
                NanopubLoader.load(m.getNanopub(), m.getCounter());
                if (m.getCounter() % 10 == 0) {
                    // Save the committed counter only every 10 nanopubs to reduce DB load
                    saveCommittedCounter(type);
                    lastSavedCounter.set(m.getCounter());
                }
                lastCommittedCounter = m.getCounter();
                loaded.getAndIncrement();

                if (loaded.get() % 50 == 0) {
                    long currTime = System.currentTimeMillis();
                    double speed = 50 / ((currTime - checkpointTime.get()) / 1000.0);
                    log.info("Loading speed: " + String.format("%.2f", speed) +
                            " np/s. Counter: " + lastCommittedCounter);
                    checkpointTime.set(currTime);
                    checkpointCounter.set(lastCommittedCounter);
                }
            });
            // Make sure to save the last committed counter at the end of the batch
            if (lastCommittedCounter >= lastSavedCounter.get()) {
                saveCommittedCounter(type);
            }
        } catch (IOException e) {
            throw new RuntimeException("I/O error while reading the response Jelly stream.", e);
        } finally {
            try {
                response.close();
            } catch (IOException e) {
                log.info("Failed to close the Jelly stream response.");
            }
        }
    }

    /**
     * Save the last committed counter to the DB. Do this every N nanopubs to reduce DB load.
     * Remember to call this method at the end of the batch as well.
     *
     * @param type the type of loading operation (initial or update)
     */
    private static void saveCommittedCounter(LoadingType type) {
        try {
            if (type == LoadingType.INITIAL) {
                StatusController.get().setLoadingInitial(lastCommittedCounter);
            } else {
                StatusController.get().setLoadingUpdates(lastCommittedCounter);
            }
        } catch (Exception e) {
            throw new RuntimeException("Could not update the nanopub counter in DB", e);
        }
    }

    /**
     * Set the last known setup ID. Called from MainVerticle on startup to restore persisted state.
     *
     * @param setupId the setup ID to set, or null if not known
     */
    static void setLastKnownSetupId(Long setupId) {
        lastKnownSetupId = setupId;
    }

    /**
     * Update the cached metadata fields used for forwarding to clients.
     */
    private static void updateForwardingMetadata(RegistryMetadata metadata) {
        lastCoverageTypes = metadata.coverageTypes();
        lastCoverageAgents = metadata.coverageAgents();
        lastTestInstance = metadata.testInstance();
        lastNanopubCount = metadata.nanopubCount();
    }

    /**
     * Run a HEAD request to the Registry to fetch its current metadata (load counter and setup ID).
     *
     * @return the registry metadata
     */
    static RegistryMetadata fetchRegistryMetadata() {
        int tries = 0;
        RegistryMetadata metadata = null;
        while (metadata == null && tries < MAX_RETRIES_METADATA) {
            try {
                metadata = fetchRegistryMetadataInner();
            } catch (Exception e) {
                tries++;
                log.info("Failed to fetch registry metadata, try " + tries +
                        ". Retrying in {}ms...", RETRY_DELAY_METADATA);
                log.info("Failure Reason: ", e);
                try {
                    Thread.sleep(RETRY_DELAY_METADATA);
                } catch (InterruptedException e2) {
                    throw new RuntimeException(
                            "Interrupted while waiting to retry fetching registry metadata.");
                }
            }
        }
        if (metadata == null) {
            throw new RuntimeException("Failed to fetch registry metadata after " +
                    MAX_RETRIES_METADATA + " retries.");
        }
        return metadata;
    }

    /**
     * Inner logic for fetching the registry metadata via HEAD request.
     *
     * @return the registry metadata (load counter and setup ID)
     * @throws IOException if the HTTP request fails
     */
    private static RegistryMetadata fetchRegistryMetadataInner() throws IOException {
        var request = new HttpHead(registryUrl);
        try (var response = metadataClient.execute(request)) {
            int status = response.getStatusLine().getStatusCode();
            EntityUtils.consumeQuietly(response.getEntity());
            if (status < 200 || status >= 300) {
                throw new RuntimeException("Registry metadata HTTP status is not 2xx: " +
                        status + ".");
            }

            // Check if the registry is ready
            var hStatus = response.getHeaders("Nanopub-Registry-Status");
            if (hStatus.length == 0) {
                throw new RuntimeException("Registry did not return a Nanopub-Registry-Status header.");
            }
            if (!"ready".equals(hStatus[0].getValue()) && !"updating".equals(hStatus[0].getValue())) {
                throw new RuntimeException("Registry is not in ready state.");
            }

            // Get the load counter
            var hCounter = response.getHeaders("Nanopub-Registry-Load-Counter");
            if (hCounter.length == 0) {
                throw new RuntimeException("Registry did not return a Nanopub-Registry-Load-Counter header.");
            }
            long loadCounter = Long.parseLong(hCounter[0].getValue());

            // Get the setup ID (optional — older registries may not have it)
            Long setupId = null;
            var hSetupId = response.getHeaders("Nanopub-Registry-Setup-Id");
            if (hSetupId.length > 0) {
                try {
                    setupId = Long.parseLong(hSetupId[0].getValue());
                } catch (NumberFormatException e) {
                    log.info("Could not parse Nanopub-Registry-Setup-Id header: {}", hSetupId[0].getValue());
                }
            }

            // Read metadata headers for forwarding to clients
            String coverageTypes = getHeaderValue(response, "Nanopub-Registry-Coverage-Types");
            String coverageAgents = getHeaderValue(response, "Nanopub-Registry-Coverage-Agents");
            String testInstance = getHeaderValue(response, "Nanopub-Registry-Test-Instance");
            String nanopubCount = getHeaderValue(response, "Nanopub-Registry-Nanopub-Count");
            // Optional — older registries (without trust calculation) won't set this header.
            String trustStateHash = getHeaderValue(response, "Nanopub-Registry-Trust-State-Hash");

            return new RegistryMetadata(loadCounter, setupId, coverageTypes, coverageAgents,
                    testInstance, nanopubCount, trustStateHash);
        }
    }

    private static String getHeaderValue(CloseableHttpResponse response, String name) {
        var headers = response.getHeaders(name);
        return headers.length > 0 ? headers[0].getValue() : null;
    }

    /**
     * Construct the URL for fetching the Jelly stream.
     *
     * @param afterCounter the last known counter to have been committed in the DB
     * @return the URL for fetching the Jelly stream
     */
    private static String makeStreamFetchUrl(long afterCounter) {
        return registryUrl + "nanopubs.jelly?afterCounter=" + afterCounter;
    }
}

1	package com.knowledgepixels.query;
2
3	import java.io.IOException;
4	import java.util.concurrent.atomic.AtomicLong;
5
6	import org.apache.http.client.methods.CloseableHttpResponse;
7	import org.apache.http.client.methods.HttpGet;
8	import org.apache.http.client.methods.HttpHead;
9	import org.apache.http.impl.client.CloseableHttpClient;
10	import org.apache.http.impl.client.HttpClientBuilder;
11	import org.apache.http.util.EntityUtils;
12	import org.nanopub.NanopubUtils;
13	import org.nanopub.jelly.NanopubStream;
14	import org.slf4j.Logger;
15	import org.slf4j.LoggerFactory;
16
17	/**
18	* Loads nanopubs from the attached Nanopub Registry via a restartable Jelly stream.
19	*/
20	public class JellyNanopubLoader {	×
21	static final String registryUrl;
22	private static long lastCommittedCounter = -1;	6✔
23	private static Long lastKnownSetupId = null;	6✔
24	// Latest registry metadata fields, updated on each metadata fetch and forwarded to clients
25	static volatile String lastCoverageTypes = null;	6✔
26	static volatile String lastCoverageAgents = null;	6✔
27	static volatile String lastTestInstance = null;	6✔
28	static volatile String lastNanopubCount = null;	6✔
29	private static final CloseableHttpClient metadataClient;
30	private static final CloseableHttpClient jellyStreamClient;
31
32	private static final int MAX_RETRIES_METADATA = 10;
33	private static final int RETRY_DELAY_METADATA = 3000;
34	private static final int RETRY_DELAY_JELLY = 5000;
35
36	/**
37	* Circuit-breaker state. Counts consecutive {@code loadUpdates} invocations in
38	* which the catch block fired; resets to zero on the next successful batch.
39	* Scheduled on the single-threaded executor in {@link MainVerticle}, so a plain
40	* {@code int} is enough — no concurrency.
41	*
42	* <p>When the counter reaches {@link #BREAKER_THRESHOLD}, the next invocation
43	* sleeps {@link #BREAKER_PAUSE_MS} before proceeding. Lets the saturated RDF4J
44	* drain instead of being hammered every {@link #UPDATES_POLL_INTERVAL} ms.
45	*
46	* <p>Depends on {@link com.knowledgepixels.query.TripleStore}'s socket timeouts
47	* (change 1 of the fix plan) to turn parked commits into propagating exceptions;
48	* without them, {@code loadBatch} can park forever, the catch never fires, and
49	* this counter stays at zero.
50	*/
51	static volatile int consecutiveBatchFailures = 0;	6✔
52
53	static final int BREAKER_THRESHOLD = 3;
54	static final long BREAKER_PAUSE_MS = 30_000L;
55
56	/**
57	* Epoch-millis of the last successful {@code loadUpdates} invocation (whether it
58	* actually loaded a batch or was a "caught up, nothing to do" tick). Read by
59	* {@link MetricsCollector} to expose {@code registry.loader.last_successful_batch_age_seconds}
60	* so an operator can tell at a glance from Grafana whether an instance is still
61	* making progress. Updated on every non-exceptional return from loadUpdates,
62	* including idle returns — an instance that is caught up and correctly polling
63	* still counts as alive.
64	*/
65	static volatile long lastSuccessfulBatchAtMs = 0L;	6✔
66
67	/**
68	* Heartbeat counter for loadUpdates invocations. A summary log line is emitted
69	* every {@link #HEARTBEAT_INTERVAL_INVOCATIONS} invocations so a truncated or
70	* sampled log still shows the loader's state evolving. At the default 2 s poll
71	* interval, 30 invocations ≈ 1 line per minute.
72	*/
73	private static long loadUpdatesInvocations = 0L;	6✔
74	private static final long HEARTBEAT_INTERVAL_INVOCATIONS = 30;
75
76	private static final Logger log = LoggerFactory.getLogger(JellyNanopubLoader.class);	9✔
77
78	/**
79	* Registry metadata returned by a HEAD request.
80	*/
81	record RegistryMetadata(long loadCounter, Long setupId, String coverageTypes,	72✔
82	String coverageAgents, String testInstance, String nanopubCount,
83	String trustStateHash) {}
84
85	/**
86	* The interval in milliseconds at which the updates loader should poll for new nanopubs.
87	*/
88	public static final int UPDATES_POLL_INTERVAL = 2000;
89
90	enum LoadingType {	×
91	INITIAL,	×
92	UPDATE,	×
93	}
94
95	static {
96	// Initialize registryUrl
97	var url = Utils.getEnvString(	12✔
98	"REGISTRY_FIXED_URL", "https://registry.knowledgepixels.com/"
99	);
100	if (!url.endsWith("/")) url += "/";	12!
101	registryUrl = url;	6✔
102
103	metadataClient = HttpClientBuilder.create().setDefaultRequestConfig(Utils.getHttpRequestConfig()).build();	15✔
104	jellyStreamClient = NanopubUtils.getHttpClient();	6✔
105	}	3✔
106
107	/**
108	* Start or continue (after restart) the initial loading procedure. This simply loads all
109	* nanopubs from the attached Registry.
110	*
111	* @param afterCounter which counter to start from (-1 for the beginning)
112	*/
113	public static void loadInitial(long afterCounter) {
114	RegistryMetadata metadata = fetchRegistryMetadata();	6✔
115	updateForwardingMetadata(metadata);	6✔
116	TrustStateLoader.maybeUpdate(metadata.trustStateHash());	9✔
117	long targetCounter = metadata.loadCounter();	9✔
118	log.info("Fetched Registry load counter: {}", targetCounter);	15✔
119	// Store setupId on initial load
120	if (metadata.setupId() != null && lastKnownSetupId == null) {	9!
121	lastKnownSetupId = metadata.setupId();	×
122	StatusController.get().setRegistrySetupId(metadata.setupId());	×
123	}
124	lastCommittedCounter = afterCounter;	6✔
125	while (lastCommittedCounter < targetCounter) {	12!
126	try {
127	loadBatch(lastCommittedCounter, LoadingType.INITIAL);	×
128	log.info("Initial load: loaded batch up to counter {}", lastCommittedCounter);	×
129	} catch (Exception e) {	×
130	log.info("Failed to load batch starting from counter {}", lastCommittedCounter);	×
131	log.info("Failure reason: ", e);	×
132	try {
133	Thread.sleep(RETRY_DELAY_JELLY);	×
134	} catch (InterruptedException e2) {	×
135	throw new RuntimeException("Interrupted while waiting to retry loading batch.");	×
136	}	×
137	}	×
138	}
139	log.info("Initial load complete.");	9✔
140	}	3✔
141
142	/**
143	* Check if the Registry has any new nanopubs. If it does, load them.
144	* This method should be called periodically, and you should wait for it to finish before
145	* calling it again.
146	*/
147	public static void loadUpdates() {
148	// Circuit breaker: after BREAKER_THRESHOLD consecutive failed batches, pause
149	// before the next attempt so a saturated RDF4J can drain. Check happens before
150	// any RDF4J-touching work so the sleep isn't itself under the broken regime.
151	if (consecutiveBatchFailures >= BREAKER_THRESHOLD) {	×
152	log.warn("Circuit breaker active after {} consecutive batch failures; pausing {} ms before next attempt",	×
153	consecutiveBatchFailures, BREAKER_PAUSE_MS);	×
154	try {
155	Thread.sleep(BREAKER_PAUSE_MS);	×
156	} catch (InterruptedException e) {	×
157	// Preserve interruption semantics so a graceful shutdown (e.g. via
158	// MainVerticle's shutdown hook) isn't blocked by the pause.
159	Thread.currentThread().interrupt();	×
160	return;	×
161	}	×
162	}
163	try {
164	final var status = StatusController.get().getState();	×
165	lastCommittedCounter = status.loadCounter;	×
166	RegistryMetadata metadata = fetchRegistryMetadata();	×
167	updateForwardingMetadata(metadata);	×
168	TrustStateLoader.maybeUpdate(metadata.trustStateHash());	×
169	long targetCounter = metadata.loadCounter();	×
170	Long currentSetupId = metadata.setupId();	×
171
172	// Detect reset via setupId change
173	if (lastKnownSetupId != null && currentSetupId != null	×
174	&& !lastKnownSetupId.equals(currentSetupId)) {	×
175	log.warn("Registry reset detected: setupId {} -> {}", lastKnownSetupId, currentSetupId);	×
176	performResync(currentSetupId);	×
177	return;	×
178	}
179	// Detect reset via counter decrease (also covers first run after upgrade
180	// where no setupId was persisted yet but the registry has already been reset)
181	if (lastCommittedCounter > 0 && targetCounter >= 0	×
182	&& targetCounter < lastCommittedCounter) {
183	log.warn("Registry counter decreased {} -> {}, triggering resync",	×
184	lastCommittedCounter, targetCounter);	×
185	performResync(currentSetupId);	×
186	return;	×
187	}
188
189	// Update lastKnownSetupId on first successful poll
190	if (currentSetupId != null && lastKnownSetupId == null) {	×
191	if (lastCommittedCounter > 0) {	×
192	// Upgrade from a version without setupId tracking. The DB has data but
193	// we can't verify it matches the current registry. Force a resync.
194	log.warn("No stored setupId but DB has data (counter: {}). "	×
195	+ "Forcing resync to ensure data consistency.", lastCommittedCounter);	×
196	performResync(currentSetupId);	×
197	return;	×
198	}
199	lastKnownSetupId = currentSetupId;	×
200	StatusController.get().setRegistrySetupId(currentSetupId);	×
201	}
202
203	if (lastCommittedCounter >= targetCounter) {	×
204	// Nothing to do. Keep state at READY (setReady is idempotent) and
205	// skip the redundant setLoadingUpdates → setReady admin-repo write
206	// that the old flow did on every idle poll. Also reset the breaker
207	// counter — a successful "nothing to do" is still a successful tick
208	// and should clear stale failure state from earlier transient errors.
209	StatusController.get().setReady();	×
210	consecutiveBatchFailures = 0;	×
211	lastSuccessfulBatchAtMs = System.currentTimeMillis();	×
212	maybeLogHeartbeat(targetCounter, true);	×
213	return;	×
214	}
215	StatusController.get().setLoadingUpdates(status.loadCounter);	×
216	loadBatch(lastCommittedCounter, LoadingType.UPDATE);	×
217	// Batch completed without an exception — reset the breaker counter.
218	consecutiveBatchFailures = 0;	×
219	lastSuccessfulBatchAtMs = System.currentTimeMillis();	×
220	maybeLogHeartbeat(targetCounter, false);	×
221	log.info("Loaded {} update(s). Counter: {}, target was: {}",	×
222	lastCommittedCounter - status.loadCounter, lastCommittedCounter, targetCounter);	×
223	if (lastCommittedCounter < targetCounter) {	×
224	log.info("Warning: expected to load nanopubs up to (inclusive) counter " +	×
225	targetCounter + " based on the counter reported in Registry's headers, " +
226	"but loaded only up to {}.", lastCommittedCounter);	×
227	}
228	} catch (Exception e) {	×
229	consecutiveBatchFailures++;	×
230	log.warn("Failed to load updates. Current counter: {} (consecutive failures: {})",	×
231	lastCommittedCounter, consecutiveBatchFailures, e);	×
232	} finally {
233	try {
234	StatusController.get().setReady();	×
235	} catch (Exception e) {	×
236	log.info("Update loader: failed to set status to READY.");	×
237	log.info("Failure Reason: ", e);	×
238	}	×
239	}
240	}	×
241
242	/**
243	* Emit a heartbeat summary log line roughly every
244	* {@link #HEARTBEAT_INTERVAL_INVOCATIONS} invocations of {@link #loadUpdates}.
245	* Lets an operator reconstruct loader progress from a sparse or sampled log
246	* export, independent of Prometheus retention.
247	*/
248	private static void maybeLogHeartbeat(long targetCounter, boolean idle) {
249	loadUpdatesInvocations++;	×
250	if (loadUpdatesInvocations % HEARTBEAT_INTERVAL_INVOCATIONS != 0) return;	×
251	log.info("Loader heartbeat: counter={} target={} idle={} consecutiveBatchFailures={} breakerActive={}",	×
252	lastCommittedCounter, targetCounter, idle, consecutiveBatchFailures,	×
253	consecutiveBatchFailures >= BREAKER_THRESHOLD);	×
254	}	×
255
256	/**
257	* Re-stream all nanopubs from the registry after a reset is detected.
258	* Existing nanopubs are skipped by NanopubLoader's per-repo dedup.
259	*
260	* @param newSetupId the new setup ID from the registry, or null if unknown
261	*/
262	private static void performResync(Long newSetupId) {
263	log.warn("Starting resync with registry. New setupId: {}", newSetupId);	×
264	StatusController.get().setResetting();	×
265	lastKnownSetupId = newSetupId;	×
266	if (newSetupId != null) {	×
267	StatusController.get().setRegistrySetupId(newSetupId);	×
268	}
269	StatusController.get().setLoadingInitial(-1);	×
270	loadInitial(-1);	×
271	StatusController.get().setReady();	×
272	log.warn("Resync complete. Counter: {}", lastCommittedCounter);	×
273	}	×
274
275	/**
276	* Load a batch of nanopubs from the Jelly stream.
277	* <p>
278	* The method requests the list of all nanopubs from the Registry and reads it for as long
279	* as it can. If the stream is interrupted, the method will throw an exception, and you
280	* can resume loading from the last known counter.
281	*
282	* @param afterCounter the last known nanopub counter to have been committed in the DB
283	* @param type the type of loading operation (initial or update)
284	*/
285	static void loadBatch(long afterCounter, LoadingType type) {
286	CloseableHttpResponse response;
287	try {
288	var request = new HttpGet(makeStreamFetchUrl(afterCounter));	×
289	response = jellyStreamClient.execute(request);	×
290	} catch (IOException e) {	×
291	throw new RuntimeException("Failed to fetch Jelly stream from the Registry (I/O error).", e);	×
292	}	×
293
294	int httpStatus = response.getStatusLine().getStatusCode();	×
295	if (httpStatus < 200 \|\| httpStatus >= 300) {	×
296	EntityUtils.consumeQuietly(response.getEntity());	×
297	throw new RuntimeException("Jelly stream HTTP status is not 2xx: " + httpStatus + ".");	×
298	}
299
300	try (
301	var is = response.getEntity().getContent();	×
302	var npStream = NanopubStream.fromByteStream(is).getAsNanopubs()	×
303	) {
304	AtomicLong checkpointTime = new AtomicLong(System.currentTimeMillis());	×
305	AtomicLong checkpointCounter = new AtomicLong(lastCommittedCounter);	×
306	AtomicLong lastSavedCounter = new AtomicLong(lastCommittedCounter);	×
307	AtomicLong loaded = new AtomicLong(0L);	×
308
309	npStream.forEach(m -> {	×
310	if (!m.isSuccess()) throw new RuntimeException("Failed to load " +	×
311	"nanopub from Jelly stream. Last known counter: " + lastCommittedCounter,
312	m.getException()	×
313	);
314	if (m.getCounter() < lastCommittedCounter) {	×
315	throw new RuntimeException("Received a nanopub with a counter lower than " +	×
316	"the last known counter. Last known counter: " + lastCommittedCounter +
317	", received counter: " + m.getCounter());	×
318	}
319	NanopubLoader.load(m.getNanopub(), m.getCounter());	×
320	if (m.getCounter() % 10 == 0) {	×
321	// Save the committed counter only every 10 nanopubs to reduce DB load
322	saveCommittedCounter(type);	×
323	lastSavedCounter.set(m.getCounter());	×
324	}
325	lastCommittedCounter = m.getCounter();	×
326	loaded.getAndIncrement();	×
327
328	if (loaded.get() % 50 == 0) {	×
329	long currTime = System.currentTimeMillis();	×
330	double speed = 50 / ((currTime - checkpointTime.get()) / 1000.0);	×
331	log.info("Loading speed: " + String.format("%.2f", speed) +	×
332	" np/s. Counter: " + lastCommittedCounter);
333	checkpointTime.set(currTime);	×
334	checkpointCounter.set(lastCommittedCounter);	×
335	}
336	});	×
337	// Make sure to save the last committed counter at the end of the batch
338	if (lastCommittedCounter >= lastSavedCounter.get()) {	×
339	saveCommittedCounter(type);	×
340	}
341	} catch (IOException e) {	×
342	throw new RuntimeException("I/O error while reading the response Jelly stream.", e);	×
343	} finally {
344	try {
345	response.close();	×
346	} catch (IOException e) {	×
347	log.info("Failed to close the Jelly stream response.");	×
348	}	×
349	}
350	}	×
351
352	/**
353	* Save the last committed counter to the DB. Do this every N nanopubs to reduce DB load.
354	* Remember to call this method at the end of the batch as well.
355	*
356	* @param type the type of loading operation (initial or update)
357	*/
358	private static void saveCommittedCounter(LoadingType type) {
359	try {
360	if (type == LoadingType.INITIAL) {	×
361	StatusController.get().setLoadingInitial(lastCommittedCounter);	×
362	} else {
363	StatusController.get().setLoadingUpdates(lastCommittedCounter);	×
364	}
365	} catch (Exception e) {	×
366	throw new RuntimeException("Could not update the nanopub counter in DB", e);	×
367	}	×
368	}	×
369
370	/**
371	* Set the last known setup ID. Called from MainVerticle on startup to restore persisted state.
372	*
373	* @param setupId the setup ID to set, or null if not known
374	*/
375	static void setLastKnownSetupId(Long setupId) {
376	lastKnownSetupId = setupId;	×
377	}	×
378
379	/**
380	* Update the cached metadata fields used for forwarding to clients.
381	*/
382	private static void updateForwardingMetadata(RegistryMetadata metadata) {
383	lastCoverageTypes = metadata.coverageTypes();	9✔
384	lastCoverageAgents = metadata.coverageAgents();	9✔
385	lastTestInstance = metadata.testInstance();	9✔
386	lastNanopubCount = metadata.nanopubCount();	9✔
387	}	3✔
388
389	/**
390	* Run a HEAD request to the Registry to fetch its current metadata (load counter and setup ID).
391	*
392	* @return the registry metadata
393	*/
394	static RegistryMetadata fetchRegistryMetadata() {
395	int tries = 0;	6✔
396	RegistryMetadata metadata = null;	6✔
397	while (metadata == null && tries < MAX_RETRIES_METADATA) {	15!
398	try {
399	metadata = fetchRegistryMetadataInner();	6✔
400	} catch (Exception e) {	×
401	tries++;	×
402	log.info("Failed to fetch registry metadata, try " + tries +	×
403	". Retrying in {}ms...", RETRY_DELAY_METADATA);	×
404	log.info("Failure Reason: ", e);	×
405	try {
406	Thread.sleep(RETRY_DELAY_METADATA);	×
407	} catch (InterruptedException e2) {	×
408	throw new RuntimeException(	×
409	"Interrupted while waiting to retry fetching registry metadata.");
410	}	×
411	}	3✔
412	}
413	if (metadata == null) {	6!
414	throw new RuntimeException("Failed to fetch registry metadata after " +	×
415	MAX_RETRIES_METADATA + " retries.");
416	}
417	return metadata;	6✔
418	}
419
420	/**
421	* Inner logic for fetching the registry metadata via HEAD request.
422	*
423	* @return the registry metadata (load counter and setup ID)
424	* @throws IOException if the HTTP request fails
425	*/
426	private static RegistryMetadata fetchRegistryMetadataInner() throws IOException {
427	var request = new HttpHead(registryUrl);	15✔
428	try (var response = metadataClient.execute(request)) {	12✔
429	int status = response.getStatusLine().getStatusCode();	12✔
430	EntityUtils.consumeQuietly(response.getEntity());	9✔
431	if (status < 200 \|\| status >= 300) {	18!
432	throw new RuntimeException("Registry metadata HTTP status is not 2xx: " +	×
433	status + ".");
434	}
435
436	// Check if the registry is ready
437	var hStatus = response.getHeaders("Nanopub-Registry-Status");	12✔
438	if (hStatus.length == 0) {	9!
439	throw new RuntimeException("Registry did not return a Nanopub-Registry-Status header.");	×
440	}
441	if (!"ready".equals(hStatus[0].getValue()) && !"updating".equals(hStatus[0].getValue())) {	21!
442	throw new RuntimeException("Registry is not in ready state.");	×
443	}
444
445	// Get the load counter
446	var hCounter = response.getHeaders("Nanopub-Registry-Load-Counter");	12✔
447	if (hCounter.length == 0) {	9!
448	throw new RuntimeException("Registry did not return a Nanopub-Registry-Load-Counter header.");	×
449	}
450	long loadCounter = Long.parseLong(hCounter[0].getValue());	18✔
451
452	// Get the setup ID (optional — older registries may not have it)
453	Long setupId = null;	6✔
454	var hSetupId = response.getHeaders("Nanopub-Registry-Setup-Id");	12✔
455	if (hSetupId.length > 0) {	9!
456	try {
457	setupId = Long.parseLong(hSetupId[0].getValue());	21✔
458	} catch (NumberFormatException e) {	×
459	log.info("Could not parse Nanopub-Registry-Setup-Id header: {}", hSetupId[0].getValue());	×
460	}	3✔
461	}
462
463	// Read metadata headers for forwarding to clients
464	String coverageTypes = getHeaderValue(response, "Nanopub-Registry-Coverage-Types");	12✔
465	String coverageAgents = getHeaderValue(response, "Nanopub-Registry-Coverage-Agents");	12✔
466	String testInstance = getHeaderValue(response, "Nanopub-Registry-Test-Instance");	12✔
467	String nanopubCount = getHeaderValue(response, "Nanopub-Registry-Nanopub-Count");	12✔
468	// Optional — older registries (without trust calculation) won't set this header.
469	String trustStateHash = getHeaderValue(response, "Nanopub-Registry-Trust-State-Hash");	12✔
470
471	return new RegistryMetadata(loadCounter, setupId, coverageTypes, coverageAgents,	39✔
472	testInstance, nanopubCount, trustStateHash);
473	}
474	}
475
476	private static String getHeaderValue(CloseableHttpResponse response, String name) {
477	var headers = response.getHeaders(name);	12✔
478	return headers.length > 0 ? headers[0].getValue() : null;	27!
479	}
480
481	/**
482	* Construct the URL for fetching the Jelly stream.
483	*
484	* @param afterCounter the last known counter to have been committed in the DB
485	* @return the URL for fetching the Jelly stream
486	*/
487	private static String makeStreamFetchUrl(long afterCounter) {
488	return registryUrl + "nanopubs.jelly?afterCounter=" + afterCounter;	×
489	}
490	}

knowledgepixels / nanopub-query / 25035879209

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