24138607049

Committed 08 Apr 2026 01:42PM UTC coverage: 32.47% (-0.4%) from 32.824%

Build # 24138607049

Build Type

Pull #99

github

Committed by

web-flow

Commit Message

Merge 336546501 into 689a63b39

Pull Request Pull Request #99: Fix peer sync race with committed counter watermark

Coverage Stats

268 of 926 branches covered (28.94%)

Branch coverage included in aggregate %.

797 of 2354 relevant lines covered (33.86%)

5.7 hits per line

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

0.0

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

package com.knowledgepixels.registry;

import com.mongodb.client.ClientSession;
import io.micrometer.prometheus.PrometheusMeterRegistry;
import io.vertx.core.AbstractVerticle;
import io.vertx.core.Handler;
import io.vertx.core.Promise;
import io.vertx.core.http.HttpMethod;
import io.vertx.core.http.HttpServer;
import io.vertx.ext.web.Router;
import io.vertx.ext.web.RoutingContext;
import io.vertx.micrometer.PrometheusScrapingHandler;
import io.vertx.micrometer.backends.BackendRegistries;
import net.trustyuri.TrustyUriUtils;
import org.eclipse.rdf4j.rio.RDFFormat;
import org.eclipse.rdf4j.rio.Rio;
import org.nanopub.MalformedNanopubException;
import org.nanopub.Nanopub;
import org.nanopub.NanopubImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

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

public class MainVerticle extends AbstractVerticle {

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

    @Override
    public void start(Promise<Void> startPromise) {
        HttpServer server = vertx.createHttpServer();
        Router router = Router.router(vertx);
        server.requestHandler(router);
        server.listen(9292);
        router.route(HttpMethod.GET, "/agent*").handler(c -> {
            // /agent/... | /agents | /agentAccounts
            ListPage.show(c);
        });
        router.route(HttpMethod.GET, "/nanopubs*").handler(c -> ListPage.show(c));
        router.route(HttpMethod.GET, "/list*").handler(c -> ListPage.show(c));
        router.route(HttpMethod.GET, "/pubkeys*").handler(c -> ListPage.show(c));
        router.route(HttpMethod.GET, "/np/").handler(c -> c.response().putHeader("Location", "/").setStatusCode(307).end());
        router.route(HttpMethod.GET, "/np/*").handler(c -> NanopubPage.show(c));
        router.route(HttpMethod.GET, "/get/").handler(c -> c.response().putHeader("Location", "/").setStatusCode(307).end());
        router.route(HttpMethod.GET, "/get/*").handler(c -> NanopubPage.show(c, true));
        router.route(HttpMethod.GET, "/debug/*").handler(c -> DebugPage.show(c));
        router.route(HttpMethod.GET, "/style.css").handler(c -> ResourcePage.show(c, "style.css", "text/css"));

        // Metrics
        final var metricsHttpServer = vertx.createHttpServer();
        final var metricsRouter = Router.router(vertx);
        metricsHttpServer.requestHandler(metricsRouter).listen(9293);

        final var metricsRegistry = (PrometheusMeterRegistry) BackendRegistries.getDefaultNow();
        final var collector = new MetricsCollector(metricsRegistry);
        metricsRouter.route("/metrics").handler(PrometheusScrapingHandler.create(metricsRegistry));

        router.route(HttpMethod.GET, "/*").handler(c -> MainPage.show(c));
        router.route(HttpMethod.HEAD, "/*").handler(c -> MainPage.show(c));

        Handler<RoutingContext> postHandler = c -> {
            c.request().bodyHandler(bh -> {
                vertx.<Void>executeBlocking(() -> {
                    String contentType = c.request().getHeader("Content-Type");
                    Nanopub np = null;
                    try {
                        np = new NanopubImpl(bh.toString(), Rio.getParserFormatForMIMEType(contentType).orElse(RDFFormat.TRIG));
                    } catch (MalformedNanopubException ex) {
                        ex.printStackTrace();
                    }
                    if (np != null) {
                        try (ClientSession s = RegistryDB.getClient().startSession()) {
                            String ac = TrustyUriUtils.getArtifactCode(np.getUri().toString());
                            if (has(s, Collection.NANOPUBS.toString(), ac)) {
                                logger.info("POST: known nanopub {}", ac);
                            } else {
                                logger.info("POST: new nanopub {}", ac);

                                // Check if this nanopub's types are covered by this registry
                                if (!CoverageFilter.isCovered(np)) {
                                    throw new RuntimeException("Nanopub types not covered by this registry: " + np.getUri());
                                }

                                // Verify signature once, pass through to avoid redundant verification:
                                String pubkey = RegistryDB.getPubkey(np);
                                if (pubkey == null)
                                    throw new RuntimeException("Nanopublication not supported: " + np.getUri());

                                // Check agent/quota restrictions
                                String pubkeyHash = Utils.getHash(pubkey);
                                if (!AgentFilter.isAllowed(s, pubkeyHash))
                                    throw new RuntimeException("Pubkey not authorized on this registry: " + pubkeyHash);
                                if (AgentFilter.isOverQuota(s, pubkeyHash))
                                    throw new RuntimeException("Quota exceeded for pubkey: " + pubkeyHash);

                                // Load to nanopub store:
                                boolean success = RegistryDB.loadNanopubVerified(s, np, pubkey, null);
                                if (!success)
                                    throw new RuntimeException("Nanopublication not supported: " + np.getUri());
                                // Load to lists, if applicable:
                                NanopubLoader.simpleLoad(s, np, pubkey);
                                // Make the new nanopub immediately visible to peers:
                                RegistryDB.updateCommittedCounter(s);
                            }
                        }
                    }
                    return null;
                }).onComplete(ar -> {
                    if (ar.succeeded()) {
                        c.response().setStatusCode(201).end();
                    } else {
                        c.response().setStatusCode(400)
                                .setStatusMessage("Error processing nanopub: " + ar.cause().getMessage()).end();
                    }
                });
            });
        };
        router.route(HttpMethod.POST, "/").handler(postHandler);
        router.route(HttpMethod.POST, "/np/").handler(postHandler);

        // INIT
        vertx.executeBlocking(() -> {
            logger.info("Starting DB initialization...");
            CoverageFilter.init();
            AgentFilter.init();
            RegistryDB.init();

            new Thread(Task::runTasks).start();

            return null;
        }).onComplete(res -> logger.info("DB initialization finished"));

        // Periodic metrics update
        vertx.setPeriodic(1000, id -> collector.updateMetrics());

        // SHUTDOWN
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                logger.info("Gracefully shutting down...");
                RegistryDB.getClient().close();
                vertx.close().toCompletionStage().toCompletableFuture().get(5, TimeUnit.SECONDS);
                logger.info("Graceful shutdown completed");
            } catch (Exception ex) {
                logger.error("Graceful shutdown failed", ex);
            }
        }));

    }

}

1	package com.knowledgepixels.registry;
2
3	import com.mongodb.client.ClientSession;
4	import io.micrometer.prometheus.PrometheusMeterRegistry;
5	import io.vertx.core.AbstractVerticle;
6	import io.vertx.core.Handler;
7	import io.vertx.core.Promise;
8	import io.vertx.core.http.HttpMethod;
9	import io.vertx.core.http.HttpServer;
10	import io.vertx.ext.web.Router;
11	import io.vertx.ext.web.RoutingContext;
12	import io.vertx.micrometer.PrometheusScrapingHandler;
13	import io.vertx.micrometer.backends.BackendRegistries;
14	import net.trustyuri.TrustyUriUtils;
15	import org.eclipse.rdf4j.rio.RDFFormat;
16	import org.eclipse.rdf4j.rio.Rio;
17	import org.nanopub.MalformedNanopubException;
18	import org.nanopub.Nanopub;
19	import org.nanopub.NanopubImpl;
20	import org.slf4j.Logger;
21	import org.slf4j.LoggerFactory;
22
23	import java.util.concurrent.TimeUnit;
24
25	import static com.knowledgepixels.registry.RegistryDB.has;
26	import static com.knowledgepixels.registry.RegistryDB.isSet;
27
28	public class MainVerticle extends AbstractVerticle {	×
29
30	private final Logger logger = LoggerFactory.getLogger(MainVerticle.class);	×
31
32	@Override
33	public void start(Promise<Void> startPromise) {
34	HttpServer server = vertx.createHttpServer();	×
35	Router router = Router.router(vertx);	×
36	server.requestHandler(router);	×
37	server.listen(9292);	×
38	router.route(HttpMethod.GET, "/agent*").handler(c -> {	×
39	// /agent/... \| /agents \| /agentAccounts
40	ListPage.show(c);	×
41	});	×
42	router.route(HttpMethod.GET, "/nanopubs*").handler(c -> ListPage.show(c));	×
43	router.route(HttpMethod.GET, "/list*").handler(c -> ListPage.show(c));	×
44	router.route(HttpMethod.GET, "/pubkeys*").handler(c -> ListPage.show(c));	×
45	router.route(HttpMethod.GET, "/np/").handler(c -> c.response().putHeader("Location", "/").setStatusCode(307).end());	×
46	router.route(HttpMethod.GET, "/np/*").handler(c -> NanopubPage.show(c));	×
47	router.route(HttpMethod.GET, "/get/").handler(c -> c.response().putHeader("Location", "/").setStatusCode(307).end());	×
48	router.route(HttpMethod.GET, "/get/*").handler(c -> NanopubPage.show(c, true));	×
49	router.route(HttpMethod.GET, "/debug/*").handler(c -> DebugPage.show(c));	×
50	router.route(HttpMethod.GET, "/style.css").handler(c -> ResourcePage.show(c, "style.css", "text/css"));	×
51
52	// Metrics
53	final var metricsHttpServer = vertx.createHttpServer();	×
54	final var metricsRouter = Router.router(vertx);	×
55	metricsHttpServer.requestHandler(metricsRouter).listen(9293);	×
56
57	final var metricsRegistry = (PrometheusMeterRegistry) BackendRegistries.getDefaultNow();	×
58	final var collector = new MetricsCollector(metricsRegistry);	×
59	metricsRouter.route("/metrics").handler(PrometheusScrapingHandler.create(metricsRegistry));	×
60
61	router.route(HttpMethod.GET, "/*").handler(c -> MainPage.show(c));	×
62	router.route(HttpMethod.HEAD, "/*").handler(c -> MainPage.show(c));	×
63
64	Handler<RoutingContext> postHandler = c -> {	×
65	c.request().bodyHandler(bh -> {	×
66	vertx.<Void>executeBlocking(() -> {	×
67	String contentType = c.request().getHeader("Content-Type");	×
68	Nanopub np = null;	×
69	try {
70	np = new NanopubImpl(bh.toString(), Rio.getParserFormatForMIMEType(contentType).orElse(RDFFormat.TRIG));	×
71	} catch (MalformedNanopubException ex) {	×
72	ex.printStackTrace();	×
73	}	×
74	if (np != null) {	×
75	try (ClientSession s = RegistryDB.getClient().startSession()) {	×
76	String ac = TrustyUriUtils.getArtifactCode(np.getUri().toString());	×
77	if (has(s, Collection.NANOPUBS.toString(), ac)) {	×
78	logger.info("POST: known nanopub {}", ac);	×
79	} else {
80	logger.info("POST: new nanopub {}", ac);	×
81
82	// Check if this nanopub's types are covered by this registry
83	if (!CoverageFilter.isCovered(np)) {	×
84	throw new RuntimeException("Nanopub types not covered by this registry: " + np.getUri());	×
85	}
86
87	// Verify signature once, pass through to avoid redundant verification:
88	String pubkey = RegistryDB.getPubkey(np);	×
89	if (pubkey == null)	×
90	throw new RuntimeException("Nanopublication not supported: " + np.getUri());	×
91
92	// Check agent/quota restrictions
93	String pubkeyHash = Utils.getHash(pubkey);	×
94	if (!AgentFilter.isAllowed(s, pubkeyHash))	×
95	throw new RuntimeException("Pubkey not authorized on this registry: " + pubkeyHash);	×
96	if (AgentFilter.isOverQuota(s, pubkeyHash))	×
97	throw new RuntimeException("Quota exceeded for pubkey: " + pubkeyHash);	×
98
99	// Load to nanopub store:
100	boolean success = RegistryDB.loadNanopubVerified(s, np, pubkey, null);	×
101	if (!success)	×
102	throw new RuntimeException("Nanopublication not supported: " + np.getUri());	×
103	// Load to lists, if applicable:
104	NanopubLoader.simpleLoad(s, np, pubkey);	×
105	// Make the new nanopub immediately visible to peers:
106	RegistryDB.updateCommittedCounter(s);	×
107	}
108	}
109	}
110	return null;	×
111	}).onComplete(ar -> {	×
112	if (ar.succeeded()) {	×
113	c.response().setStatusCode(201).end();	×
114	} else {
115	c.response().setStatusCode(400)	×
116	.setStatusMessage("Error processing nanopub: " + ar.cause().getMessage()).end();	×
117	}
118	});	×
119	});	×
120	};	×
121	router.route(HttpMethod.POST, "/").handler(postHandler);	×
122	router.route(HttpMethod.POST, "/np/").handler(postHandler);	×
123
124	// INIT
125	vertx.executeBlocking(() -> {	×
126	logger.info("Starting DB initialization...");	×
127	CoverageFilter.init();	×
128	AgentFilter.init();	×
129	RegistryDB.init();	×
130
131	new Thread(Task::runTasks).start();	×
132
133	return null;	×
134	}).onComplete(res -> logger.info("DB initialization finished"));	×
135
136	// Periodic metrics update
137	vertx.setPeriodic(1000, id -> collector.updateMetrics());	×
138
139	// SHUTDOWN
140	Runtime.getRuntime().addShutdownHook(new Thread(() -> {	×
141	try {
142	logger.info("Gracefully shutting down...");	×
143	RegistryDB.getClient().close();	×
144	vertx.close().toCompletionStage().toCompletableFuture().get(5, TimeUnit.SECONDS);	×
145	logger.info("Graceful shutdown completed");	×
146	} catch (Exception ex) {	×
147	logger.error("Graceful shutdown failed", ex);	×
148	}	×
149	}));	×
150
151	}	×
152
153	}

knowledgepixels / nanopub-registry / 24138607049

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