30841151872

Committed 03 Aug 2026 06:24PM UTC coverage: 79.024% (+46.7%) from 32.285%

Build # 30841151872

Build Type

Pull #123

github

Committed by

web-flow

Commit Message

Merge 13c9ce8ff into e5ed6d462

Pull Request Pull Request #123: Add unit tests for existing functionality

Coverage Stats

808 of 1112 branches covered (72.66%)

Branch coverage included in aggregate %.

2575 of 3169 relevant lines covered (81.26%)

12.43 hits per line

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

51.82

src/main/java/com/knowledgepixels/registry/NanopubLoader.java

package com.knowledgepixels.registry;

import com.mongodb.ErrorCategory;
import com.mongodb.MongoWriteException;
import com.mongodb.client.ClientSession;
import com.mongodb.client.MongoCursor;
import net.trustyuri.TrustyUriUtils;
import net.trustyuri.rdf.RdfModule;
import org.apache.http.Header;
import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.util.EntityUtils;
import org.bson.Document;
import org.bson.types.Binary;
import org.eclipse.rdf4j.common.exception.RDF4JException;
import org.eclipse.rdf4j.rio.RDFFormat;
import org.nanopub.MalformedNanopubException;
import org.nanopub.Nanopub;
import org.nanopub.NanopubImpl;
import org.nanopub.NanopubUtils;
import org.nanopub.extra.server.GetNanopub;
import org.nanopub.jelly.JellyUtils;
import org.nanopub.jelly.MaybeNanopub;
import org.nanopub.jelly.NanopubStream;
import org.nanopub.trusty.TrustyNanopubUtils;
import org.nanopub.vocabulary.NPX;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.stream.Stream;

import static com.knowledgepixels.registry.RegistryDB.has;
import static com.knowledgepixels.registry.RegistryDB.insert;

public class NanopubLoader {

    private NanopubLoader() {
    }

    public final static String INTRO_TYPE = NPX.DECLARED_BY.stringValue();
    public final static String INTRO_TYPE_HASH = Utils.getHash(INTRO_TYPE);
    public final static String ENDORSE_TYPE = Utils.APPROVES_OF.stringValue();
    public final static String ENDORSE_TYPE_HASH = Utils.getHash(ENDORSE_TYPE);
    private static final Logger logger = LoggerFactory.getLogger(NanopubLoader.class);

    // TODO Distinguish and support these cases:
    //      1. Simple load: load to all core lists if pubkey is "core-loaded", or load to all lists if pubkey is "full-loaded"
    //      2. Core load: load to all core lists (initialize if needed), or load to all lists if pubkey is "full-loaded"
    //      3. Full load: load to all lists (initialize if needed)

    public static void simpleLoad(ClientSession mongoSession, String nanopubId) {
        simpleLoad(mongoSession, nanopubId, true);
    }

    public static void simpleLoad(ClientSession mongoSession, String nanopubId, boolean persistOnRetrieve) {
        if (persistOnRetrieve) {
            simpleLoad(mongoSession, retrieveNanopub(mongoSession, nanopubId));
        } else {
            Nanopub np = retrieveLocalNanopub(mongoSession, nanopubId);
            if (np == null) {
                logger.debug("Nanopub {} not found locally; fetching from peers without persisting on retrieve", nanopubId);
                np = getNanopub(nanopubId);
            }
            if (np != null) {
                simpleLoad(mongoSession, np);
            } else {
                logger.warn("Could not retrieve nanopub {} from any peer; skipping load", nanopubId);
            }
        }
    }

    public static void simpleLoad(ClientSession mongoSession, Nanopub np) {
        String pubkey = RegistryDB.getPubkey(np);
        if (pubkey == null) {
            logger.warn("Skipping load of nanopub {}: no valid signature found, so its public key could not be determined", np.getUri());
            return;
        }
        simpleLoad(mongoSession, np, pubkey);
    }

    /**
     * Loads a nanopub to the appropriate lists, using a pre-verified public key
     * to skip redundant signature verification.
     */
    public static void simpleLoad(ClientSession mongoSession, Nanopub np, String verifiedPubkey) {
        String pubkeyHash = Utils.getHash(verifiedPubkey);
        // TODO Do we need to load anything else here, into the other DB collections?
        if (has(mongoSession, "lists", new Document("pubkey", pubkeyHash).append("type", "$").append("status", "loaded"))) {
            logger.debug("Loading nanopub {} into full-loaded lists for pubkey {}", np.getUri(), pubkeyHash);
            RegistryDB.loadNanopubVerified(mongoSession, np, verifiedPubkey, pubkeyHash, "$");
        } else if (has(mongoSession, "lists", new Document("pubkey", pubkeyHash).append("type", INTRO_TYPE_HASH).append("status", "loaded"))) {
            logger.debug("Loading nanopub {} into core lists (intro/endorse) for pubkey {}", np.getUri(), pubkeyHash);
            RegistryDB.loadNanopubVerified(mongoSession, np, verifiedPubkey, pubkeyHash, INTRO_TYPE, ENDORSE_TYPE);
        } else {
            // Pubkey not yet loaded (unknown or in transitional "encountered" state): store the
            // nanopub in the nanopubs collection so it is not lost. RUN_OPTIONAL_LOAD will add it
            // to the appropriate lists once the pubkey's intro/endorse have been fetched.
            logger.debug("Pubkey {} not yet loaded; storing nanopub {} without adding it to any list yet", pubkeyHash, np.getUri());
            RegistryDB.loadNanopubVerified(mongoSession, np, verifiedPubkey, null);
            if (!has(mongoSession, "lists", new Document("pubkey", pubkeyHash).append("type", INTRO_TYPE_HASH))) {
                // Unknown pubkey: create encountered intro list so RUN_OPTIONAL_LOAD picks it up
                try {
                    logger.info("Encountered new pubkey {}; creating intro list entry so it can be processed by RUN_OPTIONAL_LOAD", pubkeyHash);
                    insert(mongoSession, "lists", new Document("pubkey", pubkeyHash)
                            .append("type", INTRO_TYPE_HASH)
                            .append("status", EntryStatus.encountered.getValue()));
                } catch (MongoWriteException e) {
                    if (e.getError().getCategory() != ErrorCategory.DUPLICATE_KEY) {
                        throw e;
                    }
                    logger.debug("Intro list entry for pubkey {} was already created concurrently; ignoring duplicate-key error", pubkeyHash);
                }
            }
        }
    }

    private static final int LOAD_PARALLELISM = Integer.parseInt(
            Utils.getEnv("REGISTRY_LOAD_PARALLELISM", String.valueOf(Runtime.getRuntime().availableProcessors())));

    /**
     * Processes a stream of nanopubs in parallel using a thread pool.
     * Each worker thread uses its own MongoDB ClientSession.
     * Backpressure is applied via a semaphore to avoid unbounded memory growth.
     *
     * @param stream    the nanopub stream to process
     * @param processor consumer that processes each nanopub (called with its own ClientSession)
     */
    public static void loadStreamInParallel(Stream<MaybeNanopub> stream, Consumer<Nanopub> processor) {
        if (LOAD_PARALLELISM <= 1) {
            // Fall back to sequential processing
            logger.debug("REGISTRY_LOAD_PARALLELISM={}; processing nanopub stream sequentially", LOAD_PARALLELISM);
            stream.forEach(m -> {
                if (!m.isSuccess()) {
                    logger.error("Failed to download a nanopub from the stream; aborting task");
                    throw new AbortingTaskException("Failed to download nanopub; aborting task...");
                }
                processor.accept(m.getNanopub());
            });
            return;
        }

        logger.debug("Processing nanopub stream in parallel with {} worker threads", LOAD_PARALLELISM);
        AtomicReference<Exception> error;
        try (ExecutorService executor = Executors.newFixedThreadPool(LOAD_PARALLELISM)) {
            Semaphore semaphore = new Semaphore(LOAD_PARALLELISM * 2);
            error = new AtomicReference<>();

            try {
                stream.forEach(m -> {
                    if (error.get() != null) {
                        return;
                    }
                    if (!m.isSuccess()) {
                        logger.error("Failed to download a nanopub from the stream; aborting remaining work");
                        error.compareAndSet(null, new AbortingTaskException("Failed to download nanopub; aborting task..."));
                        return;
                    }
                    Nanopub np = m.getNanopub();
                    try {
                        semaphore.acquire();
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                        logger.warn("Interrupted while waiting for a free worker slot; aborting parallel load", e);
                        error.compareAndSet(null, e);
                        return;
                    }
                    executor.submit(() -> {
                        try {
                            processor.accept(np);
                        } catch (Exception e) {
                            logger.error("Worker thread failed while processing nanopub {}: {}", np.getUri(), e.getMessage(), e);
                            error.compareAndSet(null, e);
                        } finally {
                            semaphore.release();
                        }
                    });
                });
            } finally {
                executor.shutdown();
                try {
                    if (!executor.awaitTermination(1, TimeUnit.HOURS)) {
                        logger.warn("Worker pool did not terminate within the 1-hour timeout after shutdown");
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    logger.warn("Interrupted while waiting for worker pool to terminate", e);
                }
            }
        }

        if (error.get() != null) {
            logger.error("Parallel nanopub loading failed: {}", error.get().getMessage());
            if (error.get() instanceof RuntimeException re) {
                throw re;
            }
            throw new RuntimeException("Parallel loading failed", error.get());
        }
    }

    /**
     * Retrieve Nanopubs from the peers of this Nanopub Registry.
     *
     * @param typeHash   The hash of the type of the Nanopub to retrieve.
     * @param pubkeyHash The hash of the pubkey of the Nanopub to retrieve.
     * @return A stream of MaybeNanopub objects, or an empty stream if no peer is available.
     */
    public static Stream<MaybeNanopub> retrieveNanopubsFromPeers(String typeHash, String pubkeyHash) {
        return retrieveNanopubsFromPeers(typeHash, pubkeyHash, null);
    }

    /**
     * Retrieve Nanopubs from the peers, optionally skipping ahead using checksums.
     *
     * @param typeHash       The hash of the type of the Nanopub to retrieve.
     * @param pubkeyHash     The hash of the pubkey of the Nanopub to retrieve.
     * @param afterChecksums Comma-separated checksums for skip-ahead (geometric fallback), or null for full fetch.
     * @return A stream of MaybeNanopub objects, or an empty stream if no peer is available.
     */
    public static Stream<MaybeNanopub> retrieveNanopubsFromPeers(String typeHash, String pubkeyHash, String afterChecksums) {
        // TODO Move the code of this method to nanopub-java library.

        List<String> peerUrlsToTry = new ArrayList<>(Utils.getPeerUrls());
        Collections.shuffle(peerUrlsToTry);
        if (peerUrlsToTry.isEmpty()) {
            logger.warn("No peers configured; cannot retrieve nanopub list for pubkey {} / type {}", pubkeyHash, typeHash);
        }
        while (!peerUrlsToTry.isEmpty()) {
            String peerUrl = peerUrlsToTry.removeFirst();

            String requestUrl = peerUrl + "list/" + pubkeyHash + "/" + typeHash + ".jelly";
            if (afterChecksums != null) {
                requestUrl += "?afterChecksums=" + afterChecksums;
            }
            logger.debug("Fetching nanopub list from peer: {}", requestUrl);
            try {
                CloseableHttpResponse resp = NanopubUtils.getHttpClient().execute(new HttpGet(requestUrl));
                int httpStatus = resp.getStatusLine().getStatusCode();
                if (httpStatus < 200 || httpStatus >= 300) {
                    logger.warn("Peer {} returned HTTP {} for nanopub list request {}; trying next peer", peerUrl, httpStatus, requestUrl);
                    EntityUtils.consumeQuietly(resp.getEntity());
                    continue;
                }
                Header nrStatus = resp.getFirstHeader("Nanopub-Registry-Status");
                if (nrStatus == null) {
                    logger.warn("Peer {} did not return a Nanopub-Registry-Status header for {}; trying next peer", peerUrl, requestUrl);
                    EntityUtils.consumeQuietly(resp.getEntity());
                    continue;
                } else if (!nrStatus.getValue().equals("ready") && !nrStatus.getValue().equals("updating")) {
                    logger.warn("Skipping peer {}: registry status is '{}' (expected 'ready' or 'updating'); trying next peer", peerUrl, nrStatus.getValue());
                    EntityUtils.consumeQuietly(resp.getEntity());
                    continue;
                }
                logger.debug("Successfully fetched nanopub list from peer {} (status: {})", peerUrl, nrStatus.getValue());
                InputStream is = resp.getEntity().getContent();
                return NanopubStream.fromByteStream(is).getAsNanopubs().onClose(() -> {
                    try {
                        resp.close();
                    } catch (IOException e) {
                        logger.debug("Error closing HTTP response from peer {}", peerUrl, e);
                    }
                });
            } catch (UnsupportedOperationException | IOException ex) {
                logger.warn("Failed to fetch nanopub list from peer {} ({}): {}", peerUrl, requestUrl, ex.getMessage(), ex);
            }
        }
        logger.warn("Exhausted all peers without successfully retrieving nanopub list for pubkey {} / type {}", pubkeyHash, typeHash);
        return Stream.empty();
    }

    public static Nanopub retrieveNanopub(ClientSession mongoSession, String nanopubId) {
        Nanopub np = retrieveLocalNanopub(mongoSession, nanopubId);
        int tryCount = 0;
        while (np == null) {
            if (tryCount > 10) {
                logger.error("Giving up on retrieving nanopub {} after {} attempts", nanopubId, tryCount);
                throw new RuntimeException("Could not load nanopub: " + nanopubId);
            } else if (tryCount > 0) {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException ex) {
                    logger.warn("Thread interrupted while waiting to retry nanopub retrieval for {}", nanopubId, ex);
                }
            }
            logger.info("Nanopub {} not found locally; fetching from peers (attempt {} of 10)", nanopubId, tryCount + 1);

            // TODO Reach out to other Nanopub Registries here:
            np = getNanopub(nanopubId);
            if (np != null) {
                logger.debug("Retrieved nanopub {} from a peer; persisting it locally", nanopubId);
                RegistryDB.loadNanopub(mongoSession, np);
            } else {
                logger.debug("Attempt {} to retrieve nanopub {} from peers failed", tryCount + 1, nanopubId);
                tryCount = tryCount + 1;
            }
        }
        return np;
    }

    public static Nanopub retrieveLocalNanopub(ClientSession mongoSession, String nanopubId) {
        String ac = TrustyUriUtils.getArtifactCode(nanopubId);
        MongoCursor<Document> cursor = RegistryDB.get(mongoSession, Collection.NANOPUBS.toString(), new Document("_id", ac));
        if (!cursor.hasNext()) {
            return null;
        }
        try {
            // Parse from Jelly, not TriG (it's faster)
            return JellyUtils.readFromDB(((Binary) cursor.next().get("jelly")).getData());
        } catch (RDF4JException | MalformedNanopubException ex) {
            logger.error("Failed to parse locally stored Jelly content for nanopub '{}'; treating it as missing", nanopubId, ex);
            return null;
        }
    }

    // TODO Provide this method in nanopub-java (GetNanopub)
    private static Nanopub getNanopub(String uriOrArtifactCode) {
        List<String> peerUrls = new ArrayList<>(Utils.getPeerUrls());
        Collections.shuffle(peerUrls);
        String ac = GetNanopub.getArtifactCode(uriOrArtifactCode).toString();
        if (!ac.startsWith(RdfModule.MODULE_ID)) {
            throw new IllegalArgumentException("Not a trusty URI of type RA");
        }
        if (peerUrls.isEmpty()) {
            logger.warn("No peers configured; cannot fetch nanopub {}", ac);
        }
        while (!peerUrls.isEmpty()) {
            String peerUrl = peerUrls.removeFirst();
            try {
                Nanopub np = get(ac, peerUrl, NanopubUtils.getHttpClient());
                if (np != null) {
                    logger.debug("Successfully fetched nanopub {} from peer {}", ac, peerUrl);
                    return np;
                }
            } catch (IOException | RDF4JException | MalformedNanopubException ex) {
                logger.debug("Failed to fetch nanopub {} from peer {}: {}", ac, peerUrl, ex.getMessage(), ex);
            }
        }
        logger.warn("Could not fetch nanopub {} from any of the {} configured peer(s)", ac, Utils.getPeerUrls().size());
        return null;
    }

    // TODO Provide this method in nanopub-java (GetNanopub)
    private static Nanopub get(String artifactCode, String registryUrl, HttpClient httpClient)
            throws IOException, RDF4JException, MalformedNanopubException {
        HttpGet get = null;
        // TODO Get in Jelly format:
        String getUrl = registryUrl + "np/" + artifactCode;
        try {
            get = new HttpGet(getUrl);
        } catch (IllegalArgumentException ex) {
            throw new IOException("invalid URL: " + getUrl);
        }
        get.setHeader("Accept", "application/trig");
        InputStream in = null;
        try {
            HttpResponse resp = httpClient.execute(get);
            if (!wasSuccessful(resp)) {
                EntityUtils.consumeQuietly(resp.getEntity());
                throw new IOException("Request to " + getUrl + " failed: " + resp.getStatusLine());
            }
            in = resp.getEntity().getContent();
            Nanopub nanopub = new NanopubImpl(in, RDFFormat.TRIG);
            if (!TrustyNanopubUtils.isValidTrustyNanopub(nanopub)) {
                throw new MalformedNanopubException("Nanopub retrieved from " + registryUrl + " is not a valid trusty nanopub");
            }
            return nanopub;
        } finally {
            if (in != null) {
                in.close();
            }
        }
    }

    private static boolean wasSuccessful(HttpResponse resp) {
        int c = resp.getStatusLine().getStatusCode();
        return c >= 200 && c < 300;
    }

}

1	package com.knowledgepixels.registry;
2
3	import com.mongodb.ErrorCategory;
4	import com.mongodb.MongoWriteException;
5	import com.mongodb.client.ClientSession;
6	import com.mongodb.client.MongoCursor;
7	import net.trustyuri.TrustyUriUtils;
8	import net.trustyuri.rdf.RdfModule;
9	import org.apache.http.Header;
10	import org.apache.http.HttpResponse;
11	import org.apache.http.client.HttpClient;
12	import org.apache.http.client.methods.CloseableHttpResponse;
13	import org.apache.http.client.methods.HttpGet;
14	import org.apache.http.util.EntityUtils;
15	import org.bson.Document;
16	import org.bson.types.Binary;
17	import org.eclipse.rdf4j.common.exception.RDF4JException;
18	import org.eclipse.rdf4j.rio.RDFFormat;
19	import org.nanopub.MalformedNanopubException;
20	import org.nanopub.Nanopub;
21	import org.nanopub.NanopubImpl;
22	import org.nanopub.NanopubUtils;
23	import org.nanopub.extra.server.GetNanopub;
24	import org.nanopub.jelly.JellyUtils;
25	import org.nanopub.jelly.MaybeNanopub;
26	import org.nanopub.jelly.NanopubStream;
27	import org.nanopub.trusty.TrustyNanopubUtils;
28	import org.nanopub.vocabulary.NPX;
29	import org.slf4j.Logger;
30	import org.slf4j.LoggerFactory;
31
32	import java.io.IOException;
33	import java.io.InputStream;
34	import java.util.ArrayList;
35	import java.util.Collections;
36	import java.util.List;
37	import java.util.concurrent.ExecutorService;
38	import java.util.concurrent.Executors;
39	import java.util.concurrent.Semaphore;
40	import java.util.concurrent.TimeUnit;
41	import java.util.concurrent.atomic.AtomicReference;
42	import java.util.function.Consumer;
43	import java.util.stream.Stream;
44
45	import static com.knowledgepixels.registry.RegistryDB.has;
46	import static com.knowledgepixels.registry.RegistryDB.insert;
47
48	public class NanopubLoader {
49
50	private NanopubLoader() {
51	}
52
53	public final static String INTRO_TYPE = NPX.DECLARED_BY.stringValue();	9✔
54	public final static String INTRO_TYPE_HASH = Utils.getHash(INTRO_TYPE);	9✔
55	public final static String ENDORSE_TYPE = Utils.APPROVES_OF.stringValue();	9✔
56	public final static String ENDORSE_TYPE_HASH = Utils.getHash(ENDORSE_TYPE);	9✔
57	private static final Logger logger = LoggerFactory.getLogger(NanopubLoader.class);	9✔
58
59	// TODO Distinguish and support these cases:
60	// 1. Simple load: load to all core lists if pubkey is "core-loaded", or load to all lists if pubkey is "full-loaded"
61	// 2. Core load: load to all core lists (initialize if needed), or load to all lists if pubkey is "full-loaded"
62	// 3. Full load: load to all lists (initialize if needed)
63
64	public static void simpleLoad(ClientSession mongoSession, String nanopubId) {
65	simpleLoad(mongoSession, nanopubId, true);	×
66	}	×
67
68	public static void simpleLoad(ClientSession mongoSession, String nanopubId, boolean persistOnRetrieve) {
69	if (persistOnRetrieve) {	×
70	simpleLoad(mongoSession, retrieveNanopub(mongoSession, nanopubId));	×
71	} else {
72	Nanopub np = retrieveLocalNanopub(mongoSession, nanopubId);	×
73	if (np == null) {	×
74	logger.debug("Nanopub {} not found locally; fetching from peers without persisting on retrieve", nanopubId);	×
75	np = getNanopub(nanopubId);	×
76	}
77	if (np != null) {	×
78	simpleLoad(mongoSession, np);	×
79	} else {
80	logger.warn("Could not retrieve nanopub {} from any peer; skipping load", nanopubId);	×
81	}
82	}
83	}	×
84
85	public static void simpleLoad(ClientSession mongoSession, Nanopub np) {
86	String pubkey = RegistryDB.getPubkey(np);	9✔
87	if (pubkey == null) {	6!
88	logger.warn("Skipping load of nanopub {}: no valid signature found, so its public key could not be determined", np.getUri());	15✔
89	return;	3✔
90	}
91	simpleLoad(mongoSession, np, pubkey);	×
92	}	×
93
94	/**
95	* Loads a nanopub to the appropriate lists, using a pre-verified public key
96	* to skip redundant signature verification.
97	*/
98	public static void simpleLoad(ClientSession mongoSession, Nanopub np, String verifiedPubkey) {
99	String pubkeyHash = Utils.getHash(verifiedPubkey);	9✔
100	// TODO Do we need to load anything else here, into the other DB collections?
101	if (has(mongoSession, "lists", new Document("pubkey", pubkeyHash).append("type", "$").append("status", "loaded"))) {	45✔
102	logger.debug("Loading nanopub {} into full-loaded lists for pubkey {}", np.getUri(), pubkeyHash);	18✔
103	RegistryDB.loadNanopubVerified(mongoSession, np, verifiedPubkey, pubkeyHash, "$");	39✔
104	} else if (has(mongoSession, "lists", new Document("pubkey", pubkeyHash).append("type", INTRO_TYPE_HASH).append("status", "loaded"))) {	45✔
105	logger.debug("Loading nanopub {} into core lists (intro/endorse) for pubkey {}", np.getUri(), pubkeyHash);	18✔
106	RegistryDB.loadNanopubVerified(mongoSession, np, verifiedPubkey, pubkeyHash, INTRO_TYPE, ENDORSE_TYPE);	51✔
107	} else {
108	// Pubkey not yet loaded (unknown or in transitional "encountered" state): store the
109	// nanopub in the nanopubs collection so it is not lost. RUN_OPTIONAL_LOAD will add it
110	// to the appropriate lists once the pubkey's intro/endorse have been fetched.
111	logger.debug("Pubkey {} not yet loaded; storing nanopub {} without adding it to any list yet", pubkeyHash, np.getUri());	18✔
112	RegistryDB.loadNanopubVerified(mongoSession, np, verifiedPubkey, null);	24✔
113	if (!has(mongoSession, "lists", new Document("pubkey", pubkeyHash).append("type", INTRO_TYPE_HASH))) {	36✔
114	// Unknown pubkey: create encountered intro list so RUN_OPTIONAL_LOAD picks it up
115	try {
116	logger.info("Encountered new pubkey {}; creating intro list entry so it can be processed by RUN_OPTIONAL_LOAD", pubkeyHash);	12✔
117	insert(mongoSession, "lists", new Document("pubkey", pubkeyHash)	30✔
118	.append("type", INTRO_TYPE_HASH)	9✔
119	.append("status", EntryStatus.encountered.getValue()));	6✔
120	} catch (MongoWriteException e) {	3✔
121	if (e.getError().getCategory() != ErrorCategory.DUPLICATE_KEY) {	15✔
122	throw e;	6✔
123	}
124	logger.debug("Intro list entry for pubkey {} was already created concurrently; ignoring duplicate-key error", pubkeyHash);	12✔
125	}	3✔
126	}
127	}
128	}	3✔
129
130	private static final int LOAD_PARALLELISM = Integer.parseInt(	12✔
131	Utils.getEnv("REGISTRY_LOAD_PARALLELISM", String.valueOf(Runtime.getRuntime().availableProcessors())));	12✔
132
133	/**
134	* Processes a stream of nanopubs in parallel using a thread pool.
135	* Each worker thread uses its own MongoDB ClientSession.
136	* Backpressure is applied via a semaphore to avoid unbounded memory growth.
137	*
138	* @param stream the nanopub stream to process
139	* @param processor consumer that processes each nanopub (called with its own ClientSession)
140	*/
141	public static void loadStreamInParallel(Stream<MaybeNanopub> stream, Consumer<Nanopub> processor) {
142	if (LOAD_PARALLELISM <= 1) {	9!
143	// Fall back to sequential processing
144	logger.debug("REGISTRY_LOAD_PARALLELISM={}; processing nanopub stream sequentially", LOAD_PARALLELISM);	×
145	stream.forEach(m -> {	×
146	if (!m.isSuccess()) {	×
147	logger.error("Failed to download a nanopub from the stream; aborting task");	×
148	throw new AbortingTaskException("Failed to download nanopub; aborting task...");	×
149	}
150	processor.accept(m.getNanopub());	×
151	});	×
152	return;	×
153	}
154
155	logger.debug("Processing nanopub stream in parallel with {} worker threads", LOAD_PARALLELISM);	15✔
156	AtomicReference<Exception> error;
157	try (ExecutorService executor = Executors.newFixedThreadPool(LOAD_PARALLELISM)) {	9✔
158	Semaphore semaphore = new Semaphore(LOAD_PARALLELISM * 2);	21✔
159	error = new AtomicReference<>();	12✔
160
161	try {
162	stream.forEach(m -> {	21✔
163	if (error.get() != null) {	9!
164	return;	×
165	}
166	if (!m.isSuccess()) {	9✔
167	logger.error("Failed to download a nanopub from the stream; aborting remaining work");	9✔
168	error.compareAndSet(null, new AbortingTaskException("Failed to download nanopub; aborting task..."));	24✔
169	return;	3✔
170	}
171	Nanopub np = m.getNanopub();	9✔
172	try {
173	semaphore.acquire();	6✔
174	} catch (InterruptedException e) {	×
175	Thread.currentThread().interrupt();	×
176	logger.warn("Interrupted while waiting for a free worker slot; aborting parallel load", e);	×
177	error.compareAndSet(null, e);	×
178	return;	×
179	}	3✔
180	executor.submit(() -> {	24✔
181	try {
182	processor.accept(np);	9✔
183	} catch (Exception e) {	3✔
184	logger.error("Worker thread failed while processing nanopub {}: {}", np.getUri(), e.getMessage(), e);	57✔
185	error.compareAndSet(null, e);	15✔
186	} finally {
187	semaphore.release();	6✔
188	}
189	});	3✔
190	});	3✔
191	} finally {
192	executor.shutdown();	6✔
193	try {
194	if (!executor.awaitTermination(1, TimeUnit.HOURS)) {	15!
195	logger.warn("Worker pool did not terminate within the 1-hour timeout after shutdown");	×
196	}
197	} catch (InterruptedException e) {	×
198	Thread.currentThread().interrupt();	×
199	logger.warn("Interrupted while waiting for worker pool to terminate", e);	×
200	}	3✔
201	}
202	}
203
204	if (error.get() != null) {	9✔
205	logger.error("Parallel nanopub loading failed: {}", error.get().getMessage());	21✔
206	if (error.get() instanceof RuntimeException re) {	27!
207	throw re;	6✔
208	}
209	throw new RuntimeException("Parallel loading failed", error.get());	×
210	}
211	}	3✔
212
213	/**
214	* Retrieve Nanopubs from the peers of this Nanopub Registry.
215	*
216	* @param typeHash The hash of the type of the Nanopub to retrieve.
217	* @param pubkeyHash The hash of the pubkey of the Nanopub to retrieve.
218	* @return A stream of MaybeNanopub objects, or an empty stream if no peer is available.
219	*/
220	public static Stream<MaybeNanopub> retrieveNanopubsFromPeers(String typeHash, String pubkeyHash) {
221	return retrieveNanopubsFromPeers(typeHash, pubkeyHash, null);	15✔
222	}
223
224	/**
225	* Retrieve Nanopubs from the peers, optionally skipping ahead using checksums.
226	*
227	* @param typeHash The hash of the type of the Nanopub to retrieve.
228	* @param pubkeyHash The hash of the pubkey of the Nanopub to retrieve.
229	* @param afterChecksums Comma-separated checksums for skip-ahead (geometric fallback), or null for full fetch.
230	* @return A stream of MaybeNanopub objects, or an empty stream if no peer is available.
231	*/
232	public static Stream<MaybeNanopub> retrieveNanopubsFromPeers(String typeHash, String pubkeyHash, String afterChecksums) {
233	// TODO Move the code of this method to nanopub-java library.
234
235	List<String> peerUrlsToTry = new ArrayList<>(Utils.getPeerUrls());	15✔
236	Collections.shuffle(peerUrlsToTry);	6✔
237	if (peerUrlsToTry.isEmpty()) {	9✔
238	logger.warn("No peers configured; cannot retrieve nanopub list for pubkey {} / type {}", pubkeyHash, typeHash);	15✔
239	}
240	while (!peerUrlsToTry.isEmpty()) {	9✔
241	String peerUrl = peerUrlsToTry.removeFirst();	12✔
242
243	String requestUrl = peerUrl + "list/" + pubkeyHash + "/" + typeHash + ".jelly";	15✔
244	if (afterChecksums != null) {	6✔
245	requestUrl += "?afterChecksums=" + afterChecksums;	12✔
246	}
247	logger.debug("Fetching nanopub list from peer: {}", requestUrl);	12✔
248	try {
249	CloseableHttpResponse resp = NanopubUtils.getHttpClient().execute(new HttpGet(requestUrl));	21✔
250	int httpStatus = resp.getStatusLine().getStatusCode();	12✔
251	if (httpStatus < 200 \|\| httpStatus >= 300) {	18!
252	logger.warn("Peer {} returned HTTP {} for nanopub list request {}; trying next peer", peerUrl, httpStatus, requestUrl);	54✔
253	EntityUtils.consumeQuietly(resp.getEntity());	9✔
254	continue;	3✔
255	}
256	Header nrStatus = resp.getFirstHeader("Nanopub-Registry-Status");	12✔
257	if (nrStatus == null) {	6✔
258	logger.warn("Peer {} did not return a Nanopub-Registry-Status header for {}; trying next peer", peerUrl, requestUrl);	15✔
259	EntityUtils.consumeQuietly(resp.getEntity());	9✔
260	continue;	3✔
261	} else if (!nrStatus.getValue().equals("ready") && !nrStatus.getValue().equals("updating")) {	30✔
262	logger.warn("Skipping peer {}: registry status is '{}' (expected 'ready' or 'updating'); trying next peer", peerUrl, nrStatus.getValue());	18✔
263	EntityUtils.consumeQuietly(resp.getEntity());	9✔
264	continue;	3✔
265	}
266	logger.debug("Successfully fetched nanopub list from peer {} (status: {})", peerUrl, nrStatus.getValue());	18✔
267	InputStream is = resp.getEntity().getContent();	12✔
268	return NanopubStream.fromByteStream(is).getAsNanopubs().onClose(() -> {	27✔
269	try {
270	resp.close();	×
271	} catch (IOException e) {	×
272	logger.debug("Error closing HTTP response from peer {}", peerUrl, e);	×
273	}	×
274	});	×
275	} catch (UnsupportedOperationException \| IOException ex) {	3✔
276	logger.warn("Failed to fetch nanopub list from peer {} ({}): {}", peerUrl, requestUrl, ex.getMessage(), ex);	66✔
277	}
278	}	3✔
279	logger.warn("Exhausted all peers without successfully retrieving nanopub list for pubkey {} / type {}", pubkeyHash, typeHash);	15✔
280	return Stream.empty();	6✔
281	}
282
283	public static Nanopub retrieveNanopub(ClientSession mongoSession, String nanopubId) {
284	Nanopub np = retrieveLocalNanopub(mongoSession, nanopubId);	12✔
285	int tryCount = 0;	6✔
286	while (np == null) {	6!
287	if (tryCount > 10) {	×
288	logger.error("Giving up on retrieving nanopub {} after {} attempts", nanopubId, tryCount);	×
289	throw new RuntimeException("Could not load nanopub: " + nanopubId);	×
290	} else if (tryCount > 0) {	×
291	try {
292	Thread.sleep(100);	×
293	} catch (InterruptedException ex) {	×
294	logger.warn("Thread interrupted while waiting to retry nanopub retrieval for {}", nanopubId, ex);	×
295	}	×
296	}
297	logger.info("Nanopub {} not found locally; fetching from peers (attempt {} of 10)", nanopubId, tryCount + 1);	×
298
299	// TODO Reach out to other Nanopub Registries here:
300	np = getNanopub(nanopubId);	×
301	if (np != null) {	×
302	logger.debug("Retrieved nanopub {} from a peer; persisting it locally", nanopubId);	×
303	RegistryDB.loadNanopub(mongoSession, np);	×
304	} else {
305	logger.debug("Attempt {} to retrieve nanopub {} from peers failed", tryCount + 1, nanopubId);	×
306	tryCount = tryCount + 1;	×
307	}
308	}
309	return np;	6✔
310	}
311
312	public static Nanopub retrieveLocalNanopub(ClientSession mongoSession, String nanopubId) {
313	String ac = TrustyUriUtils.getArtifactCode(nanopubId);	9✔
314	MongoCursor<Document> cursor = RegistryDB.get(mongoSession, Collection.NANOPUBS.toString(), new Document("_id", ac));	30✔
315	if (!cursor.hasNext()) {	9✔
316	return null;	6✔
317	}
318	try {
319	// Parse from Jelly, not TriG (it's faster)
320	return JellyUtils.readFromDB(((Binary) cursor.next().get("jelly")).getData());	27✔
321	} catch (RDF4JException \| MalformedNanopubException ex) {	3✔
322	logger.error("Failed to parse locally stored Jelly content for nanopub '{}'; treating it as missing", nanopubId, ex);	15✔
323	return null;	6✔
324	}
325	}
326
327	// TODO Provide this method in nanopub-java (GetNanopub)
328	private static Nanopub getNanopub(String uriOrArtifactCode) {
329	List<String> peerUrls = new ArrayList<>(Utils.getPeerUrls());	×
330	Collections.shuffle(peerUrls);	×
331	String ac = GetNanopub.getArtifactCode(uriOrArtifactCode).toString();	×
332	if (!ac.startsWith(RdfModule.MODULE_ID)) {	×
333	throw new IllegalArgumentException("Not a trusty URI of type RA");	×
334	}
335	if (peerUrls.isEmpty()) {	×
336	logger.warn("No peers configured; cannot fetch nanopub {}", ac);	×
337	}
338	while (!peerUrls.isEmpty()) {	×
339	String peerUrl = peerUrls.removeFirst();	×
340	try {
341	Nanopub np = get(ac, peerUrl, NanopubUtils.getHttpClient());	×
342	if (np != null) {	×
343	logger.debug("Successfully fetched nanopub {} from peer {}", ac, peerUrl);	×
344	return np;	×
345	}
346	} catch (IOException \| RDF4JException \| MalformedNanopubException ex) {	×
347	logger.debug("Failed to fetch nanopub {} from peer {}: {}", ac, peerUrl, ex.getMessage(), ex);	×
348	}	×
349	}	×
350	logger.warn("Could not fetch nanopub {} from any of the {} configured peer(s)", ac, Utils.getPeerUrls().size());	×
351	return null;	×
352	}
353
354	// TODO Provide this method in nanopub-java (GetNanopub)
355	private static Nanopub get(String artifactCode, String registryUrl, HttpClient httpClient)
356	throws IOException, RDF4JException, MalformedNanopubException {
357	HttpGet get = null;	×
358	// TODO Get in Jelly format:
359	String getUrl = registryUrl + "np/" + artifactCode;	×
360	try {
361	get = new HttpGet(getUrl);	×
362	} catch (IllegalArgumentException ex) {	×
363	throw new IOException("invalid URL: " + getUrl);	×
364	}	×
365	get.setHeader("Accept", "application/trig");	×
366	InputStream in = null;	×
367	try {
368	HttpResponse resp = httpClient.execute(get);	×
369	if (!wasSuccessful(resp)) {	×
370	EntityUtils.consumeQuietly(resp.getEntity());	×
371	throw new IOException("Request to " + getUrl + " failed: " + resp.getStatusLine());	×
372	}
373	in = resp.getEntity().getContent();	×
374	Nanopub nanopub = new NanopubImpl(in, RDFFormat.TRIG);	×
375	if (!TrustyNanopubUtils.isValidTrustyNanopub(nanopub)) {	×
376	throw new MalformedNanopubException("Nanopub retrieved from " + registryUrl + " is not a valid trusty nanopub");	×
377	}
378	return nanopub;	×
379	} finally {
380	if (in != null) {	×
381	in.close();	×
382	}
383	}
384	}
385
386	private static boolean wasSuccessful(HttpResponse resp) {
387	int c = resp.getStatusLine().getStatusCode();	×
388	return c >= 200 && c < 300;	×
389	}
390
391	}

knowledgepixels / nanopub-registry / 30841151872

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