30988770240

Committed 05 Aug 2026 08:22AM UTC coverage: 61.54% (-0.07%) from 61.613%

Build # 30988770240

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #160 from knowledgepixels/fix/loader-headers-during-resync

fix(loader): keep the liveness and registry-count headers honest during a resync

Coverage Stats

663 of 1222 branches covered (54.26%)

Branch coverage included in aggregate %.

1950 of 3024 relevant lines covered (64.48%)

9.82 hits per line

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

94.07

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

package com.knowledgepixels.query;

import io.micrometer.core.instrument.Gauge;
import io.micrometer.core.instrument.MeterRegistry;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Class to collect metrics for performance analysis.
 */
public final class MetricsCollector {

    private final AtomicInteger loadCounter = new AtomicInteger(0);
    private final AtomicInteger typeRepositoriesCounter = new AtomicInteger(0);
    private final AtomicInteger pubkeyRepositoriesCounter = new AtomicInteger(0);
    private final AtomicInteger fullRepositoriesCounter = new AtomicInteger(0);

    /**
     * Value behind {@code registry.loader.sync_lag_nanopubs}. Refreshed on the
     * {@code updateMetrics} tick rather than inside the gauge lambda: the gauge is
     * evaluated on the scrape path, which runs on a Vert.x event loop, and
     * {@link NanopubLoader#getLoadedNanopubCount()} can fall through to a SPARQL
     * query on a cold cache. The repo-name counters above are populated the same
     * way and for the same reason.
     */
    private final AtomicLong syncLagNanopubs = new AtomicLong(UNKNOWN_LAG);

    /**
     * Sentinel for {@code registry.loader.sync_lag_nanopubs} when either side of the
     * subtraction is unavailable — before the first registry poll, or if the
     * forwarded count is unparseable. Distinct from {@code 0}, which asserts the
     * instance is genuinely in sync. Real lags are clamped at zero so this value
     * can never arise from arithmetic.
     */
    private static final long UNKNOWN_LAG = -1L;

    private final Map<StatusController.State, AtomicInteger> statusStates = new ConcurrentHashMap<>();

    /**
     * Creates new metrics collector object.
     *
     * @param meterRegistry The registry instance
     */
    public MetricsCollector(MeterRegistry meterRegistry) {
        // Numeric metrics
        Gauge.builder("registry.load.counter", loadCounter, AtomicInteger::get).register(meterRegistry);
        Gauge.builder("registry.type.repositories.counter", typeRepositoriesCounter, AtomicInteger::get).register(meterRegistry);
        Gauge.builder("registry.pubkey.repositories.counter", pubkeyRepositoriesCounter, AtomicInteger::get).register(meterRegistry);
        Gauge.builder("registry.full.repositories.counter", fullRepositoriesCounter, AtomicInteger::get).register(meterRegistry);

        // Circuit-breaker observability: expose both the raw counter and a boolean
        // "breaker active" flag. The boolean is redundant with counter >= threshold
        // but much cleaner to visualise in Grafana (the counter can saturate well
        // above the threshold during a sustained outage, which makes a single
        // "is the breaker tripped?" alert awkward to express over the raw value).
        Gauge.builder("registry.loader.consecutive_batch_failures",
                        () -> (double) JellyNanopubLoader.consecutiveBatchFailures)
                .description("Consecutive loadUpdates batches that threw an exception before succeeding")
                .register(meterRegistry);
        Gauge.builder("registry.loader.breaker_active",
                        () -> JellyNanopubLoader.consecutiveBatchFailures >= JellyNanopubLoader.BREAKER_THRESHOLD ? 1.0 : 0.0)
                .description("1 if the loader circuit breaker is tripped (consecutive failures >= threshold), 0 otherwise")
                .register(meterRegistry);
        // Liveness signal that works without log access: seconds since the loader last
        // demonstrably reached the triple store — a committed load counter, a completed
        // batch (initial or update), or a verified idle-tick probe. An instance whose
        // value climbs unbounded while peers stay low is stuck on something the other
        // gauges don't capture, including during an initial load or resync, which this
        // gauge could not distinguish from a stall until those paths were stamped too.
        Gauge.builder("registry.loader.last_successful_batch_age_seconds",
                        () -> {
                            long t = JellyNanopubLoader.lastSuccessfulBatchAtMs;
                            if (t == 0L) return 0.0;    // not started yet
                            return (System.currentTimeMillis() - t) / 1000.0;
                        })
                .description("Seconds since the loader last reached the store (initial load, update batch, or idle probe)")
                .register(meterRegistry);
        // How far behind its own registry this instance is. The gauges above all
        // describe the loader's *internal* health; this one is the outcome an
        // operator actually cares about, and it is absolute rather than relative —
        // unlike the monitor's cross-instance checksum comparison, it still fires
        // when every instance stalls at once (incident 2026-07-31).
        //
        // Pair it with last_successful_batch_age_seconds when alerting. The registry
        // side of the subtraction is the count from the last metadata fetch — an update
        // poll, or a mid-load refresh during a long initial load — so if fetching itself
        // is what broke, both counts freeze together and the lag reads a falsely
        // reassuring 0. Neither signal covers the other's blind spot.
        Gauge.builder("registry.loader.sync_lag_nanopubs", syncLagNanopubs, AtomicLong::get)
                .description("Nanopubs this instance is behind its registry; -1 when either count is unknown")
                .register(meterRegistry);

        // Shard-reconciliation observability (issue #139). Both read volatile
        // counters kept by ShardReconciler — no SPARQL on the scrape path. Any
        // non-zero repaired value means the backend acknowledged a shard write
        // that was not durable, and deserves an alert.
        Gauge.builder("registry.reconciler.nanopubs_checked_total",
                        () -> (double) ShardReconciler.checkedNanopubCount)
                .description("Nanopubs whose shard fan-out was verified by the reconciliation sweep since process start")
                .register(meterRegistry);
        Gauge.builder("registry.reconciler.shards_repaired_total",
                        () -> (double) ShardReconciler.repairedShardCount)
                .description("Missing shard repos detected and re-loaded by the reconciliation sweep since process start")
                .register(meterRegistry);
        Gauge.builder("registry.reconciler.shards_relost_total",
                        () -> (double) ShardReconciler.relostShardCount)
                .description("Shards that a previous sweep verified present and that later vanished (backend revoked readable state, issue #142)")
                .register(meterRegistry);

        // Status label metrics
        for (final var status : StatusController.State.values()) {
            AtomicInteger stateGauge = new AtomicInteger(0);
            statusStates.put(status, stateGauge);
            Gauge.builder("registry.server.status", stateGauge, AtomicInteger::get)
                    .description("Server status (1 if current)")
                    .tag("status", status.name())
                    .register(meterRegistry);
        }

        // Spaces / AuthorityResolver gauges. These read volatile fields kept
        // by AuthorityResolver — no SPARQL on the scrape path. Each lambda
        // re-fetches the singleton to match the lazy-init pattern used by
        // the rest of the codebase.
        Gauge.builder("registry.spaces.subjects.admin_ris",
                        () -> (double) AuthorityResolver.get().getLastSubjectTotals().adminRIs())
                .description("Distinct admin gen:RoleInstantiation subjects in the current space-state graph (last build/cycle observation)")
                .register(meterRegistry);
        Gauge.builder("registry.spaces.subjects.attachment_ras",
                        () -> (double) AuthorityResolver.get().getLastSubjectTotals().attachmentRAs())
                .description("Distinct gen:RoleAssignment subjects in the current space-state graph (last build/cycle observation)")
                .register(meterRegistry);
        Gauge.builder("registry.spaces.subjects.non_admin_ris",
                        () -> (double) AuthorityResolver.get().getLastSubjectTotals().nonAdminRIs())
                .description("Distinct non-admin gen:RoleInstantiation subjects in the current space-state graph (last build/cycle observation)")
                .register(meterRegistry);
        Gauge.builder("registry.spaces.delta.last_inserted_triples",
                        () -> (double) AuthorityResolver.get().getLastInsertedTriplesTotal())
                .description("Total inserted triples across all five tiers in the most recent full build or incremental cycle")
                .register(meterRegistry);
        Gauge.builder("registry.spaces.rebuild.last_duration_seconds",
                        () -> AuthorityResolver.get().getLastFullBuildDurationMs() / 1000.0)
                .description("Wall-clock duration of the most recent full space-state build")
                .register(meterRegistry);
        Gauge.builder("registry.spaces.cycle.last_duration_seconds",
                        () -> AuthorityResolver.get().getLastIncrementalCycleDurationMs() / 1000.0)
                .description("Wall-clock duration of the most recent incremental space-state cycle that did work")
                .register(meterRegistry);
        Gauge.builder("registry.spaces.processed_up_to_lag",
                        () -> (double) AuthorityResolver.get().getLastProcessedUpToLag())
                .description("currentLoadCounter - processedUpTo observed at the start of the most recent incremental cycle (0 after a full build)")
                .register(meterRegistry);
    }

    /**
     * Updates the metrics based on the current state of the system.
     */
    public void updateMetrics() {
        // Update numeric metrics
        loadCounter.set((int) StatusController.get().getState().loadCounter);
        // Request repository names once, to avoid multiple calls
        var repoNames = TripleStore.get().getRepositoryNames();
        if (repoNames == null) {
            repoNames = Set.of();
        }
        typeRepositoriesCounter.set(
                (int) repoNames
                        .stream()
                        .filter(repo -> repo.startsWith("type_"))
                        .count()
        );
        pubkeyRepositoriesCounter.set(
                (int) repoNames
                        .stream()
                        .filter(repo -> repo.startsWith("pubkey_"))
                        .count()
        );
        fullRepositoriesCounter.set(repoNames.size());
        // Keeps the loaded-count/checksum caches warm for applyGlobalHeaders, which
        // runs on the event loop and therefore reads them without a store fallback.
        // This tick is the right host: it already runs unconditionally on its own
        // executor at a fixed cadence, and computeSyncLag below needs the count anyway.
        NanopubLoader.primeHeaderCaches();
        syncLagNanopubs.set(computeSyncLag());

        // Update status gauge
        final var currentStatus = StatusController.get().getState().state;
        for (final var status : StatusController.State.values()) {
            statusStates.get(status).set(status.equals(currentStatus) ? 1 : 0);
        }
    }

    /**
     * Nanopubs this instance is behind its registry, or {@link #UNKNOWN_LAG} if either
     * count is unavailable.
     *
     * <p>Clamped at zero: the loaded count is bumped as each nanopub lands while the
     * registry count only refreshes once per poll, so the loaded side can legitimately
     * run ahead for a tick. Reporting that as negative would collide with the
     * unknown sentinel.
     *
     * @return the lag in nanopubs, clamped to zero, or {@link #UNKNOWN_LAG}
     */
    static long computeSyncLag() {
        String registryCount = JellyNanopubLoader.lastNanopubCount;
        Long loaded = NanopubLoader.getLoadedNanopubCount();
        if (registryCount == null || loaded == null) {
            return UNKNOWN_LAG;
        }
        try {
            return Math.max(0L, Long.parseLong(registryCount.trim()) - loaded);
        } catch (NumberFormatException ex) {
            return UNKNOWN_LAG;
        }
    }
}

1	package com.knowledgepixels.query;
2
3	import io.micrometer.core.instrument.Gauge;
4	import io.micrometer.core.instrument.MeterRegistry;
5
6	import java.util.Map;
7	import java.util.Set;
8	import java.util.concurrent.ConcurrentHashMap;
9	import java.util.concurrent.atomic.AtomicInteger;
10	import java.util.concurrent.atomic.AtomicLong;
11
12	/**
13	* Class to collect metrics for performance analysis.
14	*/
15	public final class MetricsCollector {
16
17	private final AtomicInteger loadCounter = new AtomicInteger(0);	18✔
18	private final AtomicInteger typeRepositoriesCounter = new AtomicInteger(0);	18✔
19	private final AtomicInteger pubkeyRepositoriesCounter = new AtomicInteger(0);	18✔
20	private final AtomicInteger fullRepositoriesCounter = new AtomicInteger(0);	18✔
21
22	/**
23	* Value behind {@code registry.loader.sync_lag_nanopubs}. Refreshed on the
24	* {@code updateMetrics} tick rather than inside the gauge lambda: the gauge is
25	* evaluated on the scrape path, which runs on a Vert.x event loop, and
26	* {@link NanopubLoader#getLoadedNanopubCount()} can fall through to a SPARQL
27	* query on a cold cache. The repo-name counters above are populated the same
28	* way and for the same reason.
29	*/
30	private final AtomicLong syncLagNanopubs = new AtomicLong(UNKNOWN_LAG);	18✔
31
32	/**
33	* Sentinel for {@code registry.loader.sync_lag_nanopubs} when either side of the
34	* subtraction is unavailable — before the first registry poll, or if the
35	* forwarded count is unparseable. Distinct from {@code 0}, which asserts the
36	* instance is genuinely in sync. Real lags are clamped at zero so this value
37	* can never arise from arithmetic.
38	*/
39	private static final long UNKNOWN_LAG = -1L;
40
41	private final Map<StatusController.State, AtomicInteger> statusStates = new ConcurrentHashMap<>();	15✔
42
43	/**
44	* Creates new metrics collector object.
45	*
46	* @param meterRegistry The registry instance
47	*/
48	public MetricsCollector(MeterRegistry meterRegistry) {	6✔
49	// Numeric metrics
50	Gauge.builder("registry.load.counter", loadCounter, AtomicInteger::get).register(meterRegistry);	24✔
51	Gauge.builder("registry.type.repositories.counter", typeRepositoriesCounter, AtomicInteger::get).register(meterRegistry);	24✔
52	Gauge.builder("registry.pubkey.repositories.counter", pubkeyRepositoriesCounter, AtomicInteger::get).register(meterRegistry);	24✔
53	Gauge.builder("registry.full.repositories.counter", fullRepositoriesCounter, AtomicInteger::get).register(meterRegistry);	24✔
54
55	// Circuit-breaker observability: expose both the raw counter and a boolean
56	// "breaker active" flag. The boolean is redundant with counter >= threshold
57	// but much cleaner to visualise in Grafana (the counter can saturate well
58	// above the threshold during a sustained outage, which makes a single
59	// "is the breaker tripped?" alert awkward to express over the raw value).
60	Gauge.builder("registry.loader.consecutive_batch_failures",	12✔
61	() -> (double) JellyNanopubLoader.consecutiveBatchFailures)	12✔
62	.description("Consecutive loadUpdates batches that threw an exception before succeeding")	6✔
63	.register(meterRegistry);	6✔
64	Gauge.builder("registry.loader.breaker_active",	12✔
65	() -> JellyNanopubLoader.consecutiveBatchFailures >= JellyNanopubLoader.BREAKER_THRESHOLD ? 1.0 : 0.0)	18!
66	.description("1 if the loader circuit breaker is tripped (consecutive failures >= threshold), 0 otherwise")	6✔
67	.register(meterRegistry);	6✔
68	// Liveness signal that works without log access: seconds since the loader last
69	// demonstrably reached the triple store — a committed load counter, a completed
70	// batch (initial or update), or a verified idle-tick probe. An instance whose
71	// value climbs unbounded while peers stay low is stuck on something the other
72	// gauges don't capture, including during an initial load or resync, which this
73	// gauge could not distinguish from a stall until those paths were stamped too.
74	Gauge.builder("registry.loader.last_successful_batch_age_seconds",	12✔
75	() -> {
76	long t = JellyNanopubLoader.lastSuccessfulBatchAtMs;	6✔
77	if (t == 0L) return 0.0; // not started yet	21!
78	return (System.currentTimeMillis() - t) / 1000.0;	×
79	})
80	.description("Seconds since the loader last reached the store (initial load, update batch, or idle probe)")	6✔
81	.register(meterRegistry);	6✔
82	// How far behind its own registry this instance is. The gauges above all
83	// describe the loader's internal health; this one is the outcome an
84	// operator actually cares about, and it is absolute rather than relative —
85	// unlike the monitor's cross-instance checksum comparison, it still fires
86	// when every instance stalls at once (incident 2026-07-31).
87	//
88	// Pair it with last_successful_batch_age_seconds when alerting. The registry
89	// side of the subtraction is the count from the last metadata fetch — an update
90	// poll, or a mid-load refresh during a long initial load — so if fetching itself
91	// is what broke, both counts freeze together and the lag reads a falsely
92	// reassuring 0. Neither signal covers the other's blind spot.
93	Gauge.builder("registry.loader.sync_lag_nanopubs", syncLagNanopubs, AtomicLong::get)	18✔
94	.description("Nanopubs this instance is behind its registry; -1 when either count is unknown")	6✔
95	.register(meterRegistry);	6✔
96
97	// Shard-reconciliation observability (issue #139). Both read volatile
98	// counters kept by ShardReconciler — no SPARQL on the scrape path. Any
99	// non-zero repaired value means the backend acknowledged a shard write
100	// that was not durable, and deserves an alert.
101	Gauge.builder("registry.reconciler.nanopubs_checked_total",	12✔
102	() -> (double) ShardReconciler.checkedNanopubCount)	12✔
103	.description("Nanopubs whose shard fan-out was verified by the reconciliation sweep since process start")	6✔
104	.register(meterRegistry);	6✔
105	Gauge.builder("registry.reconciler.shards_repaired_total",	12✔
106	() -> (double) ShardReconciler.repairedShardCount)	12✔
107	.description("Missing shard repos detected and re-loaded by the reconciliation sweep since process start")	6✔
108	.register(meterRegistry);	6✔
109	Gauge.builder("registry.reconciler.shards_relost_total",	12✔
110	() -> (double) ShardReconciler.relostShardCount)	12✔
111	.description("Shards that a previous sweep verified present and that later vanished (backend revoked readable state, issue #142)")	6✔
112	.register(meterRegistry);	6✔
113
114	// Status label metrics
115	for (final var status : StatusController.State.values()) {	48✔
116	AtomicInteger stateGauge = new AtomicInteger(0);	15✔
117	statusStates.put(status, stateGauge);	18✔
118	Gauge.builder("registry.server.status", stateGauge, AtomicInteger::get)	15✔
119	.description("Server status (1 if current)")	9✔
120	.tag("status", status.name())	9✔
121	.register(meterRegistry);	6✔
122	}
123
124	// Spaces / AuthorityResolver gauges. These read volatile fields kept
125	// by AuthorityResolver — no SPARQL on the scrape path. Each lambda
126	// re-fetches the singleton to match the lazy-init pattern used by
127	// the rest of the codebase.
128	Gauge.builder("registry.spaces.subjects.admin_ris",	12✔
129	() -> (double) AuthorityResolver.get().getLastSubjectTotals().adminRIs())	18✔
130	.description("Distinct admin gen:RoleInstantiation subjects in the current space-state graph (last build/cycle observation)")	6✔
131	.register(meterRegistry);	6✔
132	Gauge.builder("registry.spaces.subjects.attachment_ras",	12✔
133	() -> (double) AuthorityResolver.get().getLastSubjectTotals().attachmentRAs())	18✔
134	.description("Distinct gen:RoleAssignment subjects in the current space-state graph (last build/cycle observation)")	6✔
135	.register(meterRegistry);	6✔
136	Gauge.builder("registry.spaces.subjects.non_admin_ris",	12✔
137	() -> (double) AuthorityResolver.get().getLastSubjectTotals().nonAdminRIs())	18✔
138	.description("Distinct non-admin gen:RoleInstantiation subjects in the current space-state graph (last build/cycle observation)")	6✔
139	.register(meterRegistry);	6✔
140	Gauge.builder("registry.spaces.delta.last_inserted_triples",	12✔
141	() -> (double) AuthorityResolver.get().getLastInsertedTriplesTotal())	15✔
142	.description("Total inserted triples across all five tiers in the most recent full build or incremental cycle")	6✔
143	.register(meterRegistry);	6✔
144	Gauge.builder("registry.spaces.rebuild.last_duration_seconds",	12✔
145	() -> AuthorityResolver.get().getLastFullBuildDurationMs() / 1000.0)	21✔
146	.description("Wall-clock duration of the most recent full space-state build")	6✔
147	.register(meterRegistry);	6✔
148	Gauge.builder("registry.spaces.cycle.last_duration_seconds",	12✔
149	() -> AuthorityResolver.get().getLastIncrementalCycleDurationMs() / 1000.0)	21✔
150	.description("Wall-clock duration of the most recent incremental space-state cycle that did work")	6✔
151	.register(meterRegistry);	6✔
152	Gauge.builder("registry.spaces.processed_up_to_lag",	12✔
153	() -> (double) AuthorityResolver.get().getLastProcessedUpToLag())	15✔
154	.description("currentLoadCounter - processedUpTo observed at the start of the most recent incremental cycle (0 after a full build)")	6✔
155	.register(meterRegistry);	6✔
156	}	3✔
157
158	/**
159	* Updates the metrics based on the current state of the system.
160	*/
161	public void updateMetrics() {
162	// Update numeric metrics
163	loadCounter.set((int) StatusController.get().getState().loadCounter);	21✔
164	// Request repository names once, to avoid multiple calls
165	var repoNames = TripleStore.get().getRepositoryNames();	9✔
166	if (repoNames == null) {	6!
167	repoNames = Set.of();	×
168	}
169	typeRepositoriesCounter.set(	12✔
170	(int) repoNames
171	.stream()	6✔
172	.filter(repo -> repo.startsWith("type_"))	15✔
173	.count()	6✔
174	);
175	pubkeyRepositoriesCounter.set(	12✔
176	(int) repoNames
177	.stream()	6✔
178	.filter(repo -> repo.startsWith("pubkey_"))	15✔
179	.count()	6✔
180	);
181	fullRepositoriesCounter.set(repoNames.size());	15✔
182	// Keeps the loaded-count/checksum caches warm for applyGlobalHeaders, which
183	// runs on the event loop and therefore reads them without a store fallback.
184	// This tick is the right host: it already runs unconditionally on its own
185	// executor at a fixed cadence, and computeSyncLag below needs the count anyway.
186	NanopubLoader.primeHeaderCaches();	3✔
187	syncLagNanopubs.set(computeSyncLag());	12✔
188
189	// Update status gauge
190	final var currentStatus = StatusController.get().getState().state;	12✔
191	for (final var status : StatusController.State.values()) {	48✔
192	statusStates.get(status).set(status.equals(currentStatus) ? 1 : 0);	33!
193	}
194	}	3✔
195
196	/**
197	* Nanopubs this instance is behind its registry, or {@link #UNKNOWN_LAG} if either
198	* count is unavailable.
199	*
200	* <p>Clamped at zero: the loaded count is bumped as each nanopub lands while the
201	* registry count only refreshes once per poll, so the loaded side can legitimately
202	* run ahead for a tick. Reporting that as negative would collide with the
203	* unknown sentinel.
204	*
205	* @return the lag in nanopubs, clamped to zero, or {@link #UNKNOWN_LAG}
206	*/
207	static long computeSyncLag() {
208	String registryCount = JellyNanopubLoader.lastNanopubCount;	6✔
209	Long loaded = NanopubLoader.getLoadedNanopubCount();	6✔
210	if (registryCount == null \|\| loaded == null) {	12!
211	return UNKNOWN_LAG;	6✔
212	}
213	try {
214	return Math.max(0L, Long.parseLong(registryCount.trim()) - loaded);	27✔
215	} catch (NumberFormatException ex) {	3✔
216	return UNKNOWN_LAG;	6✔
217	}
218	}
219	}

knowledgepixels / nanopub-query / 30988770240

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