28113618611

Committed 24 Jun 2026 04:28PM UTC coverage: 31.926% (-0.2%) from 32.089%

Build # 28113618611

Build Type

Pull #116

github

Committed by

web-flow

Commit Message

Merge d931f8afc into eebd16ba4

Pull Request Pull Request #116: Enhance and standardize logging across multiple components

Coverage Stats

313 of 1106 branches covered (28.3%)

Branch coverage included in aggregate %.

1048 of 3157 relevant lines covered (33.2%)

5.51 hits per line

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

32.47

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

package com.knowledgepixels.registry;

import com.mongodb.ErrorCategory;
import com.mongodb.MongoWriteException;
import com.mongodb.client.ClientSession;
import com.mongodb.client.MongoCursor;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpHead;
import org.apache.http.util.EntityUtils;
import org.bson.Document;
import org.nanopub.NanopubUtils;
import org.nanopub.jelly.NanopubStream;
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.atomic.AtomicLong;

import static com.knowledgepixels.registry.RegistryDB.*;

/**
 * Checks peer Nanopub Registries for new nanopublications and loads them.
 */
public class RegistryPeerConnector {

    private RegistryPeerConnector() {}

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

    public static void checkPeers(ClientSession s) {
        List<String> peerUrls = new ArrayList<>(Utils.getPeerUrls());
        Collections.shuffle(peerUrls);

        for (String peerUrl : peerUrls) {
            try {
                checkPeer(s, peerUrl);
            } catch (Exception ex) {
                logger.warn("Failed to check peer {}: {} ({})", peerUrl, ex.getMessage(), ex.getClass().getSimpleName(), ex);
            }
        }
    }

    static void checkPeer(ClientSession s, String peerUrl) throws IOException {
        logger.info("Checking peer: {}", peerUrl);

        HttpResponse resp = NanopubUtils.getHttpClient().execute(new HttpHead(peerUrl));
        int httpStatus = resp.getStatusLine().getStatusCode();
        String reason = resp.getStatusLine().getReasonPhrase();
        EntityUtils.consumeQuietly(resp.getEntity());
        if (httpStatus < 200 || httpStatus >= 300) {
            logger.warn("Failed to reach peer {}: HTTP {} {} ; skipping", peerUrl, httpStatus, reason);
            return;
        }

        if (isTestInstance(resp)) {
            logger.info("Skipping peer {} because it is a test instance", peerUrl);
            return;
        }

        String status = getHeader(resp, "Nanopub-Registry-Status");
        if (!"ready".equals(status) && !"updating".equals(status)) {
            logger.warn("Skipping peer {}: registry status is '{}' (expected 'ready' or 'updating')", peerUrl, status);
            return;
        }

        String setupHeader = getHeader(resp, "Nanopub-Registry-Setup-Id");
        String loadCounterHeader = getHeader(resp, "Nanopub-Registry-Load-Counter");
        Long peerSetupId = getHeaderLong(resp, "Nanopub-Registry-Setup-Id");
        Long peerLoadCounter = getHeaderLong(resp, "Nanopub-Registry-Load-Counter");
        if (peerSetupId == null || peerLoadCounter == null) {
            logger.warn("Skipping peer {}: missing or invalid headers. Nanopub-Registry-Setup-Id='{}', Nanopub-Registry-Load-Counter='{}'", peerUrl, setupHeader, loadCounterHeader);
            return;
        }

        syncWithPeer(s, peerUrl, peerSetupId, peerLoadCounter);
    }

    static void syncWithPeer(ClientSession s, String peerUrl, long peerSetupId, long peerLoadCounter) {
        Document peerState = getPeerState(s, peerUrl);
        Long lastSetupId = peerState != null ? peerState.getLong("setupId") : null;
        Long lastLoadCounter = peerState != null ? peerState.getLong("loadCounter") : null;

        if (lastSetupId != null && !lastSetupId.equals(peerSetupId)) {
            logger.info("Peer {} was reset: setupId changed from {} -> {}; resetting tracking state", peerUrl, lastSetupId, peerSetupId);
            deletePeerState(s, peerUrl);
            lastLoadCounter = null;
        }

        long effectiveCounter = lastLoadCounter != null ? lastLoadCounter : 0;

        if (lastLoadCounter != null && lastLoadCounter.equals(peerLoadCounter)) {
            logger.info("Peer {} has no new nanopubs (loadCounter unchanged: {})", peerUrl, peerLoadCounter);
        } else if (lastLoadCounter != null) {
            // Fetch all nanopubs added since our last known position.
            logger.info("Peer {} has new nanopubs (loadCounter {} -> {}), fetching recent", peerUrl, lastLoadCounter, peerLoadCounter);
            long lastReceived = loadRecentNanopubs(s, peerUrl, lastLoadCounter);
            if (lastReceived > 0) {
                effectiveCounter = lastReceived;
                logger.info("Updated effective counter for {} to {}", peerUrl, effectiveCounter);
            } else {
                logger.info("No nanopubs were successfully received from {} when fetching recent entries", peerUrl);
            }
            // Only discover new pubkeys when the peer has new data
            discoverPubkeys(s, peerUrl);
        } else {
            logger.info("Peer {} is new to this registry; starting pubkey discovery and initial sync", peerUrl);
            discoverPubkeys(s, peerUrl);
        }
        updatePeerState(s, peerUrl, peerSetupId, effectiveCounter);
        logger.debug("Peer {} state updated: setupId={}, loadCounter={}", peerUrl, peerSetupId, effectiveCounter);
    }

    /**
     * Fetches nanopubs from a peer after the given counter.
     *
     * @return the counter of the last successfully received nanopub, or -1 if none were received
     */
    private static long loadRecentNanopubs(ClientSession s, String peerUrl, long afterCounter) {
        String requestUrl = peerUrl + "nanopubs.jelly?afterCounter=" + afterCounter;
        logger.info("Fetching recent nanopubs from {} (afterCounter={})", peerUrl, afterCounter);
        AtomicLong lastReceivedCounter = new AtomicLong(-1);
        try {
            HttpResponse resp = NanopubUtils.getHttpClient().execute(new HttpGet(requestUrl));
            int httpStatus = resp.getStatusLine().getStatusCode();
            String reason = resp.getStatusLine().getReasonPhrase();
            if (httpStatus < 200 || httpStatus >= 300) {
                EntityUtils.consumeQuietly(resp.getEntity());
                logger.warn("Fetching recent nanopubs from {} failed: HTTP {} {} ; skipping", requestUrl, httpStatus, reason);
                return -1;
            }
            try (InputStream is = resp.getEntity().getContent()) {
                NanopubLoader.loadStreamInParallel(
                        NanopubStream.fromByteStream(is).getAsNanopubs().peek(m -> {
                            // Track counter in the main thread as items are consumed from the stream
                            if (m.isSuccess() && m.getCounter() > 0) {
                                lastReceivedCounter.set(m.getCounter());
                            }
                        }),
                        np -> {
                            if (!CoverageFilter.isCovered(np)) {
                                return;
                            }
                            try (ClientSession workerSession = RegistryDB.getClient().startSession()) {
                                String pubkey = RegistryDB.getPubkey(np);
                                if (pubkey != null) {
                                    NanopubLoader.simpleLoad(workerSession, np, pubkey);
                                }
                            }
                        });
            }
        } catch (IOException ex) {
            logger.warn("Failed to fetch recent nanopubs from {} (request: {}): {} ({})", peerUrl, requestUrl, ex.getMessage(), ex.getClass().getSimpleName(), ex);
        }
        logger.info("Last received counter from {}: {}", peerUrl, lastReceivedCounter.get());
        return lastReceivedCounter.get();
    }

    static void discoverPubkeys(ClientSession s, String peerUrl) {
        logger.info("Discovering pubkeys from peer: {}", peerUrl);
        try {
            List<String> peerPubkeys = Utils.retrieveListFromJsonUrl(peerUrl + "pubkeys.json");
            int discovered = 0;
            logger.debug("Retrieved {} pubkeys from {} for discovery", peerPubkeys.size(), peerUrl);
            for (String pubkeyHash : peerPubkeys) {
                Document filter = new Document("pubkey", pubkeyHash).append("type", NanopubLoader.INTRO_TYPE_HASH);
                if (!has(s, "lists", filter)) {
                    try {
                        insert(s, "lists", new Document("pubkey", pubkeyHash)
                                .append("type", NanopubLoader.INTRO_TYPE_HASH)
                                .append("status", EntryStatus.encountered.getValue()));
                    } catch (MongoWriteException e) {
                        if (e.getError().getCategory() != ErrorCategory.DUPLICATE_KEY) {
                            throw e;
                        } else {
                            logger.debug("Pubkey {} was inserted concurrently by another worker", pubkeyHash);
                        }
                    }
                    discovered++;
                } else if (!has(s, "lists", new Document(filter).append("status", EntryStatus.loaded.getValue()))) {
                    // Set status to encountered if not already loaded (fixes null-status entries from older code)
                    collection("lists").updateMany(s, filter,
                            new Document("$set", new Document("status", EntryStatus.encountered.getValue())));
                    discovered++;
                }
            }
            logger.info("Discovered {} new pubkeys from peer {}", discovered, peerUrl);
        } catch (Exception ex) {
            logger.warn("Failed to discover pubkeys from {}: {} ({})", peerUrl, ex.getMessage(), ex.getClass().getSimpleName(), ex);
        }
    }

    static Document getPeerState(ClientSession s, String peerUrl) {
        try (MongoCursor<Document> cursor = collection(Collection.PEER_STATE.toString())
                .find(s, new Document("_id", peerUrl)).cursor()) {
            return cursor.hasNext() ? cursor.next() : null;
        }
    }

    static void updatePeerState(ClientSession s, String peerUrl, long setupId, long loadCounter) {
        collection(Collection.PEER_STATE.toString()).updateOne(s,
                new Document("_id", peerUrl),
                new Document("$set", new Document("_id", peerUrl)
                        .append("setupId", setupId)
                        .append("loadCounter", loadCounter)
                        .append("lastChecked", System.currentTimeMillis())),
                new com.mongodb.client.model.UpdateOptions().upsert(true));
    }

    static void deletePeerState(ClientSession s, String peerUrl) {
        collection(Collection.PEER_STATE.toString()).deleteOne(s, new Document("_id", peerUrl));
    }

    static boolean isTestInstance(HttpResponse resp) {
        return "true".equals(getHeader(resp, "Nanopub-Registry-Test-Instance"));
    }

    static String getHeader(HttpResponse resp, String name) {
        return resp.getFirstHeader(name) != null ? resp.getFirstHeader(name).getValue() : null;
    }

    static Long getHeaderLong(HttpResponse resp, String name) {
        String value = getHeader(resp, name);
        if (value == null || "null".equals(value)) {
            return null;
        }
        try {
            return Long.parseLong(value);
        } catch (NumberFormatException ex) {
            logger.debug("Failed to parse header {} value '{}' as long", name, value);
            return null;
        }
    }

}

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 org.apache.http.HttpResponse;
8	import org.apache.http.client.methods.HttpGet;
9	import org.apache.http.client.methods.HttpHead;
10	import org.apache.http.util.EntityUtils;
11	import org.bson.Document;
12	import org.nanopub.NanopubUtils;
13	import org.nanopub.jelly.NanopubStream;
14	import org.slf4j.Logger;
15	import org.slf4j.LoggerFactory;
16
17	import java.io.IOException;
18	import java.io.InputStream;
19	import java.util.ArrayList;
20	import java.util.Collections;
21	import java.util.List;
22	import java.util.concurrent.atomic.AtomicLong;
23
24	import static com.knowledgepixels.registry.RegistryDB.*;
25
26	/**
27	* Checks peer Nanopub Registries for new nanopublications and loads them.
28	*/
29	public class RegistryPeerConnector {
30
31	private RegistryPeerConnector() {}
32
33	private static final Logger logger = LoggerFactory.getLogger(RegistryPeerConnector.class);	12✔
34
35	public static void checkPeers(ClientSession s) {
36	List<String> peerUrls = new ArrayList<>(Utils.getPeerUrls());	×
37	Collections.shuffle(peerUrls);	×
38
39	for (String peerUrl : peerUrls) {	×
40	try {
41	checkPeer(s, peerUrl);	×
42	} catch (Exception ex) {	×
43	logger.warn("Failed to check peer {}: {} ({})", peerUrl, ex.getMessage(), ex.getClass().getSimpleName(), ex);	×
44	}	×
45	}	×
46	}	×
47
48	static void checkPeer(ClientSession s, String peerUrl) throws IOException {
49	logger.info("Checking peer: {}", peerUrl);	×
50
51	HttpResponse resp = NanopubUtils.getHttpClient().execute(new HttpHead(peerUrl));	×
52	int httpStatus = resp.getStatusLine().getStatusCode();	×
53	String reason = resp.getStatusLine().getReasonPhrase();	×
54	EntityUtils.consumeQuietly(resp.getEntity());	×
55	if (httpStatus < 200 \|\| httpStatus >= 300) {	×
56	logger.warn("Failed to reach peer {}: HTTP {} {} ; skipping", peerUrl, httpStatus, reason);	×
57	return;	×
58	}
59
60	if (isTestInstance(resp)) {	×
61	logger.info("Skipping peer {} because it is a test instance", peerUrl);	×
62	return;	×
63	}
64
65	String status = getHeader(resp, "Nanopub-Registry-Status");	×
66	if (!"ready".equals(status) && !"updating".equals(status)) {	×
67	logger.warn("Skipping peer {}: registry status is '{}' (expected 'ready' or 'updating')", peerUrl, status);	×
68	return;	×
69	}
70
71	String setupHeader = getHeader(resp, "Nanopub-Registry-Setup-Id");	×
72	String loadCounterHeader = getHeader(resp, "Nanopub-Registry-Load-Counter");	×
73	Long peerSetupId = getHeaderLong(resp, "Nanopub-Registry-Setup-Id");	×
74	Long peerLoadCounter = getHeaderLong(resp, "Nanopub-Registry-Load-Counter");	×
75	if (peerSetupId == null \|\| peerLoadCounter == null) {	×
76	logger.warn("Skipping peer {}: missing or invalid headers. Nanopub-Registry-Setup-Id='{}', Nanopub-Registry-Load-Counter='{}'", peerUrl, setupHeader, loadCounterHeader);	×
77	return;	×
78	}
79
80	syncWithPeer(s, peerUrl, peerSetupId, peerLoadCounter);	×
81	}	×
82
83	static void syncWithPeer(ClientSession s, String peerUrl, long peerSetupId, long peerLoadCounter) {
84	Document peerState = getPeerState(s, peerUrl);	12✔
85	Long lastSetupId = peerState != null ? peerState.getLong("setupId") : null;	24✔
86	Long lastLoadCounter = peerState != null ? peerState.getLong("loadCounter") : null;	24✔
87
88	if (lastSetupId != null && !lastSetupId.equals(peerSetupId)) {	21✔
89	logger.info("Peer {} was reset: setupId changed from {} -> {}; resetting tracking state", peerUrl, lastSetupId, peerSetupId);	54✔
90	deletePeerState(s, peerUrl);	9✔
91	lastLoadCounter = null;	6✔
92	}
93
94	long effectiveCounter = lastLoadCounter != null ? lastLoadCounter : 0;	21✔
95
96	if (lastLoadCounter != null && lastLoadCounter.equals(peerLoadCounter)) {	21!
97	logger.info("Peer {} has no new nanopubs (loadCounter unchanged: {})", peerUrl, peerLoadCounter);	21✔
98	} else if (lastLoadCounter != null) {	6!
99	// Fetch all nanopubs added since our last known position.
100	logger.info("Peer {} has new nanopubs (loadCounter {} -> {}), fetching recent", peerUrl, lastLoadCounter, peerLoadCounter);	×
101	long lastReceived = loadRecentNanopubs(s, peerUrl, lastLoadCounter);	×
102	if (lastReceived > 0) {	×
103	effectiveCounter = lastReceived;	×
104	logger.info("Updated effective counter for {} to {}", peerUrl, effectiveCounter);	×
105	} else {
106	logger.info("No nanopubs were successfully received from {} when fetching recent entries", peerUrl);	×
107	}
108	// Only discover new pubkeys when the peer has new data
109	discoverPubkeys(s, peerUrl);	×
110	} else {	×
111	logger.info("Peer {} is new to this registry; starting pubkey discovery and initial sync", peerUrl);	12✔
112	discoverPubkeys(s, peerUrl);	9✔
113	}
114	updatePeerState(s, peerUrl, peerSetupId, effectiveCounter);	15✔
115	logger.debug("Peer {} state updated: setupId={}, loadCounter={}", peerUrl, peerSetupId, effectiveCounter);	57✔
116	}	3✔
117
118	/**
119	* Fetches nanopubs from a peer after the given counter.
120	*
121	* @return the counter of the last successfully received nanopub, or -1 if none were received
122	*/
123	private static long loadRecentNanopubs(ClientSession s, String peerUrl, long afterCounter) {
124	String requestUrl = peerUrl + "nanopubs.jelly?afterCounter=" + afterCounter;	×
125	logger.info("Fetching recent nanopubs from {} (afterCounter={})", peerUrl, afterCounter);	×
126	AtomicLong lastReceivedCounter = new AtomicLong(-1);	×
127	try {
128	HttpResponse resp = NanopubUtils.getHttpClient().execute(new HttpGet(requestUrl));	×
129	int httpStatus = resp.getStatusLine().getStatusCode();	×
130	String reason = resp.getStatusLine().getReasonPhrase();	×
131	if (httpStatus < 200 \|\| httpStatus >= 300) {	×
132	EntityUtils.consumeQuietly(resp.getEntity());	×
133	logger.warn("Fetching recent nanopubs from {} failed: HTTP {} {} ; skipping", requestUrl, httpStatus, reason);	×
134	return -1;	×
135	}
136	try (InputStream is = resp.getEntity().getContent()) {	×
137	NanopubLoader.loadStreamInParallel(	×
138	NanopubStream.fromByteStream(is).getAsNanopubs().peek(m -> {	×
139	// Track counter in the main thread as items are consumed from the stream
140	if (m.isSuccess() && m.getCounter() > 0) {	×
141	lastReceivedCounter.set(m.getCounter());	×
142	}
143	}),	×
144	np -> {
145	if (!CoverageFilter.isCovered(np)) {	×
146	return;	×
147	}
148	try (ClientSession workerSession = RegistryDB.getClient().startSession()) {	×
149	String pubkey = RegistryDB.getPubkey(np);	×
150	if (pubkey != null) {	×
151	NanopubLoader.simpleLoad(workerSession, np, pubkey);	×
152	}
153	}
154	});	×
155	}
156	} catch (IOException ex) {	×
157	logger.warn("Failed to fetch recent nanopubs from {} (request: {}): {} ({})", peerUrl, requestUrl, ex.getMessage(), ex.getClass().getSimpleName(), ex);	×
158	}	×
159	logger.info("Last received counter from {}: {}", peerUrl, lastReceivedCounter.get());	×
160	return lastReceivedCounter.get();	×
161	}
162
163	static void discoverPubkeys(ClientSession s, String peerUrl) {
164	logger.info("Discovering pubkeys from peer: {}", peerUrl);	12✔
165	try {
166	List<String> peerPubkeys = Utils.retrieveListFromJsonUrl(peerUrl + "pubkeys.json");	×
167	int discovered = 0;	×
168	logger.debug("Retrieved {} pubkeys from {} for discovery", peerPubkeys.size(), peerUrl);	×
169	for (String pubkeyHash : peerPubkeys) {	×
170	Document filter = new Document("pubkey", pubkeyHash).append("type", NanopubLoader.INTRO_TYPE_HASH);	×
171	if (!has(s, "lists", filter)) {	×
172	try {
173	insert(s, "lists", new Document("pubkey", pubkeyHash)	×
174	.append("type", NanopubLoader.INTRO_TYPE_HASH)	×
175	.append("status", EntryStatus.encountered.getValue()));	×
176	} catch (MongoWriteException e) {	×
177	if (e.getError().getCategory() != ErrorCategory.DUPLICATE_KEY) {	×
178	throw e;	×
179	} else {
180	logger.debug("Pubkey {} was inserted concurrently by another worker", pubkeyHash);	×
181	}
182	}	×
183	discovered++;	×
184	} else if (!has(s, "lists", new Document(filter).append("status", EntryStatus.loaded.getValue()))) {	×
185	// Set status to encountered if not already loaded (fixes null-status entries from older code)
186	collection("lists").updateMany(s, filter,	×
187	new Document("$set", new Document("status", EntryStatus.encountered.getValue())));	×
188	discovered++;	×
189	}
190	}	×
191	logger.info("Discovered {} new pubkeys from peer {}", discovered, peerUrl);	×
192	} catch (Exception ex) {	3✔
193	logger.warn("Failed to discover pubkeys from {}: {} ({})", peerUrl, ex.getMessage(), ex.getClass().getSimpleName(), ex);	72✔
194	}	×
195	}	3✔
196
197	static Document getPeerState(ClientSession s, String peerUrl) {
198	try (MongoCursor<Document> cursor = collection(Collection.PEER_STATE.toString())	27✔
199	.find(s, new Document("_id", peerUrl)).cursor()) {	9✔
200	return cursor.hasNext() ? cursor.next() : null;	33✔
201	}
202	}
203
204	static void updatePeerState(ClientSession s, String peerUrl, long setupId, long loadCounter) {
205	collection(Collection.PEER_STATE.toString()).updateOne(s,	63✔
206	new Document("_id", peerUrl),
207	new Document("$set", new Document("_id", peerUrl)
208	.append("setupId", setupId)	12✔
209	.append("loadCounter", loadCounter)	9✔
210	.append("lastChecked", System.currentTimeMillis())),	24✔
211	new com.mongodb.client.model.UpdateOptions().upsert(true));	3✔
212	}	3✔
213
214	static void deletePeerState(ClientSession s, String peerUrl) {
215	collection(Collection.PEER_STATE.toString()).deleteOne(s, new Document("_id", peerUrl));	33✔
216	}	3✔
217
218	static boolean isTestInstance(HttpResponse resp) {
219	return "true".equals(getHeader(resp, "Nanopub-Registry-Test-Instance"));	18✔
220	}
221
222	static String getHeader(HttpResponse resp, String name) {
223	return resp.getFirstHeader(name) != null ? resp.getFirstHeader(name).getValue() : null;	33✔
224	}
225
226	static Long getHeaderLong(HttpResponse resp, String name) {
227	String value = getHeader(resp, name);	12✔
228	if (value == null \|\| "null".equals(value)) {	18✔
229	return null;	6✔
230	}
231	try {
232	return Long.parseLong(value);	12✔
233	} catch (NumberFormatException ex) {	3✔
234	logger.debug("Failed to parse header {} value '{}' as long", name, value);	15✔
235	return null;	6✔
236	}
237	}
238
239	}

knowledgepixels / nanopub-registry / 28113618611

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