30629636856

Committed 31 Jul 2026 12:10PM UTC coverage: 58.504% (-0.04%) from 58.54%

Build # 30629636856

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #147 from knowledgepixels/fix/status-controller-connection-lifecycle

fix(status): open a fresh admin-repo connection per transaction

Coverage Stats

621 of 1198 branches covered (51.84%)

Branch coverage included in aggregate %.

1804 of 2947 relevant lines covered (61.21%)

9.42 hits per line

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

87.72

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

package com.knowledgepixels.query;

import org.eclipse.rdf4j.common.transaction.IsolationLevels;
import org.eclipse.rdf4j.model.IRI;
import org.eclipse.rdf4j.model.Literal;
import org.eclipse.rdf4j.model.ValueFactory;
import org.eclipse.rdf4j.model.impl.SimpleValueFactory;
import org.eclipse.rdf4j.repository.RepositoryConnection;
import org.nanopub.vocabulary.NPA;

import java.util.Objects;

/**
 * Class to control the load status of the database.
 */
public class StatusController {

    /**
     * The load states in which the database can be.
     */
    public enum State {
        /**
         * The service is launching.
         */
        LAUNCHING,
        /**
         * The service is loading.
         */
        LOADING_INITIAL,
        /**
         * The service is loading updates.
         */
        LOADING_UPDATES,
        /**
         * The service is ready to serve requests.
         */
        READY,
        /**
         * The service detected a registry reset and is about to resync.
         */
        RESETTING,
    }

    /**
     * Get the singleton instance of the StatusController.
     *
     * @return the StatusController instance
     */
    public static StatusController get() {
        return instance;
    }

    private final static StatusController instance = new StatusController();

    static final IRI HAS_REGISTRY_SETUP_ID =
            SimpleValueFactory.getInstance().createIRI(NPA.NAMESPACE, "hasRegistrySetupId");

    private static final ValueFactory vf = SimpleValueFactory.getInstance();

    private boolean initialized = false;
    private volatile State state = null;
    private volatile long lastCommittedCounter = -1;
    private volatile Long registrySetupId = null;

    /**
     * Opens a fresh admin-repo connection for a single transaction.
     *
     * <p>This class used to hold one connection for the lifetime of the process,
     * assigned once in {@link #initialize()} and never re-acquired. An RDF4J restart
     * — which the {@code rdf4j} healthcheck performs by design when its probes fail —
     * orphaned that connection permanently: every subsequent admin-repo write threw,
     * so {@code setLoadingUpdates} failed before {@code loadBatch} could run and the
     * instance ingested nothing until the query container itself was restarted.
     * RDF4J recovering on its own was not enough. Acquiring per transaction means a
     * restart costs one failed tick instead of wedging the loader (incident
     * 2026-07-31, query.knowledgepixels.com).
     *
     * <p>Cheap enough to do per call: {@code getConnection()} on an
     * {@code HTTPRepository} is local and does no HTTP, as
     * {@link TripleStore#getRepoConnection(String)} notes. Every other admin-repo
     * caller in the codebase already works this way.
     *
     * @return a new connection the caller must close
     */
    private static RepositoryConnection openAdminConnection() {
        return TripleStore.get().getAdminRepoConnection();
    }

    /**
     * Rolls back an active transaction, suppressing any failure to do so.
     *
     * <p>The transaction may no longer exist server-side (already committed, timed
     * out, or lost to a restart), and that secondary failure must not mask the
     * original one.
     *
     * @param conn the connection whose transaction should be rolled back
     */
    private static void rollbackQuietly(RepositoryConnection conn) {
        try {
            if (conn.isActive()) {
                conn.rollback();
            }
        } catch (Exception rollbackException) {
            // Deliberately ignored; the caller rethrows the original cause.
        }
    }

    /**
     * Represents the current status of the service, including the load counter.
     */
    public static class LoadingStatus {

        /**
         * The current state of the service.
         */
        public final State state;

        /**
         * The current load counter.
         */
        public final long loadCounter;

        private LoadingStatus(State state, long loadCounter) {
            this.state = state;
            this.loadCounter = loadCounter;
        }

        /**
         * Create a new LoadingStatus instance.
         *
         * @param state       the current state of the service
         * @param loadCounter the current load counter
         * @return a new LoadingStatus instance
         */
        public static LoadingStatus of(State state, long loadCounter) {
            return new LoadingStatus(state, loadCounter);
        }

        @Override
        public boolean equals(Object o) {
            if (o == null || getClass() != o.getClass()) {
                return false;
            }
            LoadingStatus that = (LoadingStatus) o;
            return loadCounter == that.loadCounter && state == that.state;
        }

        @Override
        public int hashCode() {
            return Objects.hash(state, loadCounter);
        }

    }

    /**
     * Initialize the StatusController, fetching the last known state from the DB.
     * This must be called right after service startup, before loading any nanopubs.
     *
     * @return the current state and the last committed counter
     */
    public LoadingStatus initialize() {
        synchronized (this) {
            if (initialized) {
                throw new IllegalStateException("Already initialized");
            }
            state = State.LAUNCHING;
            try (RepositoryConnection conn = openAdminConnection()) {
                try {
                    // Serializable, as the service state needs to be strictly consistent
                    conn.begin(IsolationLevels.SERIALIZABLE);
                    // Fetch the state from the DB
                    try (var statements = conn.getStatements(NPA.THIS_REPO, NPA.HAS_STATUS, null, NPA.GRAPH)) {
                        if (!statements.hasNext()) {
                            conn.add(NPA.THIS_REPO, NPA.HAS_STATUS, stateAsLiteral(state), NPA.GRAPH);
                        } else {
                            var stateStatement = statements.next();
                            state = State.valueOf(stateStatement.getObject().stringValue());
                        }
                    }
                    // Fetch the load counter from the DB
                    try (var statements = conn.getStatements(NPA.THIS_REPO, NPA.HAS_REGISTRY_LOAD_COUNTER, null, NPA.GRAPH)) {
                        if (!statements.hasNext()) {
                            conn.add(NPA.THIS_REPO, NPA.HAS_REGISTRY_LOAD_COUNTER, vf.createLiteral(-1L), NPA.GRAPH);
                        } else {
                            var counterStatement = statements.next();
                            var stringVal = counterStatement.getObject().stringValue();
                            lastCommittedCounter = Long.parseLong(stringVal);
                        }
                    }
                    // Fetch the registry setup ID from the DB
                    try (var statements = conn.getStatements(NPA.THIS_REPO, HAS_REGISTRY_SETUP_ID, null, NPA.GRAPH)) {
                        if (statements.hasNext()) {
                            var setupIdStatement = statements.next();
                            registrySetupId = Long.parseLong(setupIdStatement.getObject().stringValue());
                        }
                    }
                    conn.commit();
                } catch (Exception e) {
                    rollbackQuietly(conn);
                    throw new RuntimeException(e);
                }
            }
            initialized = true;
            return getState();
        }
    }

    /**
     * Get the current state of the service.
     *
     * @return the current state and the last committed counter
     */
    public LoadingStatus getState() {
        return LoadingStatus.of(state, lastCommittedCounter);
    }

    /**
     * Transition the service to the LOADING_INITIAL state and update the load counter.
     * This should be called in two situations:
     * - By the main loading thread (after calling initialize()) to start loading the initial nanopubs.
     * - By the initial nanopub loader, as it processes the initial nanopubs.
     *
     * @param loadCounter the new load counter
     */
    public void setLoadingInitial(long loadCounter) {
        synchronized (this) {
            if (state != State.LAUNCHING && state != State.LOADING_INITIAL && state != State.RESETTING) {
                throw new IllegalStateException("Cannot transition to LOADING_INITIAL, as the " + "current state is " + state);
            }
            if (state != State.RESETTING && lastCommittedCounter > loadCounter) {
                throw new IllegalStateException("Cannot update the load counter from " + lastCommittedCounter + " to " + loadCounter);
            }
            updateState(State.LOADING_INITIAL, loadCounter);
        }
    }

    /**
     * Transition the service to the LOADING_UPDATES state and update the load counter.
     * This should be called by the updates loader, when it starts processing new nanopubs, or
     * when it finishes processing a batch of nanopubs.
     *
     * @param loadCounter the new load counter
     */
    public void setLoadingUpdates(long loadCounter) {
        synchronized (this) {
            if (state != State.LAUNCHING && state != State.LOADING_UPDATES && state != State.READY) {
                throw new IllegalStateException("Cannot transition to LOADING_UPDATES, as the " + "current state is " + state);
            }
            if (lastCommittedCounter > loadCounter) {
                throw new IllegalStateException("Cannot update the load counter from " + lastCommittedCounter + " to " + loadCounter);
            }
            // Idempotence guard: skip the admin-repo rewrite if the state and counter
            // are already where we'd set them. A no-op loop iteration of the updates
            // loader otherwise re-writes the same two triples hundreds of times an
            // hour over a long idle tail, each write growing the admin-repo LMDB via
            // copy-on-write.
            if (state == State.LOADING_UPDATES && lastCommittedCounter == loadCounter) {
                return;
            }
            updateState(State.LOADING_UPDATES, loadCounter);
        }
    }

    /**
     * Transition the service to the READY state.
     * This should be called by the loaders, after they finish their work.
     */
    public void setReady() {
        synchronized (this) {
            if (state != State.READY) {
                updateState(State.READY, lastCommittedCounter);
            }
        }
    }

    /**
     * Transition the service to the RESETTING state, resetting the load counter to -1.
     * This is triggered when a registry reset is detected (setupId changed or counter decreased),
     * or when FORCE_RESYNC is set on startup. Accepts any post-initialize source state:
     * a reset discards prior load progress, so it's valid from LOADING_INITIAL (crashed mid-init),
     * LOADING_UPDATES, READY, or RESETTING (retry after a crashed prior reset).
     */
    public void setResetting() {
        synchronized (this) {
            if (state == State.LAUNCHING) {
                throw new IllegalStateException("Cannot transition to RESETTING, as the " + "current state is " + state);
            }
            updateState(State.RESETTING, -1);
        }
    }

    /**
     * Get the persisted registry setup ID.
     *
     * @return the registry setup ID, or null if not yet known
     */
    public Long getRegistrySetupId() {
        // Lock-free read: field is volatile, writers still hold synchronized(this)
        // so there are no concurrent writers. applyGlobalHeaders in MainVerticle
        // calls this on every inbound request on the Vert.x event loop — blocking
        // here behind updateState's admin-repo transaction was a BlockedThreadChecker
        // hazard. The DB-commit-first order in the setter (setRegistrySetupId)
        // means a reader can observe the previous value for the few ms between DB
        // commit and field assignment; no caller depends on stronger consistency.
        return registrySetupId;
    }

    /**
     * Persist a new registry setup ID to the admin repo.
     *
     * @param setupId the new setup ID
     */
    public void setRegistrySetupId(long setupId) {
        synchronized (this) {
            try (RepositoryConnection conn = openAdminConnection()) {
                try {
                    conn.begin(IsolationLevels.SERIALIZABLE);
                    conn.remove(NPA.THIS_REPO, HAS_REGISTRY_SETUP_ID, null, NPA.GRAPH);
                    conn.add(NPA.THIS_REPO, HAS_REGISTRY_SETUP_ID, vf.createLiteral(setupId), NPA.GRAPH);
                    conn.commit();
                    registrySetupId = setupId;
                } catch (Exception e) {
                    rollbackQuietly(conn);
                    throw new RuntimeException(e);
                }
            }
        }
    }

    /**
     * Persist a state transition, then adopt it in memory.
     *
     * <p>The in-memory fields are assigned only after the admin-repo commit succeeds,
     * mirroring {@link #setRegistrySetupId(long)}. The earlier order (assign first,
     * commit second) left the two permanently divergent whenever the commit threw,
     * because nothing rolled the fields back: the loader would then read a
     * {@code lastCommittedCounter} ahead of the persisted one, conclude from
     * {@code lastCommittedCounter >= targetCounter} that it was caught up, and skip
     * those nanopubs forever while still reporting READY.
     *
     * <p>Readers may observe the previous state for the few ms of the commit — the
     * same window {@link #getRegistrySetupId()} already documents, and no caller
     * depends on stronger consistency. If a commit lands server-side but the response
     * is lost, the fields stay behind the DB instead of ahead of it; the next
     * transition re-writes the same triples (the remove/add pair is idempotent), so
     * that direction costs at most some re-processing rather than silent data loss.
     *
     * @param newState    the state to transition to
     * @param loadCounter the load counter to persist alongside it
     */
    void updateState(State newState, long loadCounter) {
        synchronized (this) {
            try (RepositoryConnection conn = openAdminConnection()) {
                try {
                    // Serializable, as the service state needs to be strictly consistent
                    conn.begin(IsolationLevels.SERIALIZABLE);
                    conn.remove(NPA.THIS_REPO, NPA.HAS_STATUS, null, NPA.GRAPH);
                    conn.add(NPA.THIS_REPO, NPA.HAS_STATUS, stateAsLiteral(newState), NPA.GRAPH);
                    conn.remove(NPA.THIS_REPO, NPA.HAS_REGISTRY_LOAD_COUNTER, null, NPA.GRAPH);
                    conn.add(NPA.THIS_REPO, NPA.HAS_REGISTRY_LOAD_COUNTER, vf.createLiteral(loadCounter), NPA.GRAPH);
                    conn.commit();
                    state = newState;
                    lastCommittedCounter = loadCounter;
                } catch (Exception e) {
                    rollbackQuietly(conn);
                    throw new RuntimeException(e);
                }
            }
        }
    }

    private Literal stateAsLiteral(State s) {
        return vf.createLiteral(s.toString());
    }

    /**
     * Reset the StatusController for testing purposes.
     * This will clear the state and allow re-initialization.
     */
    void resetForTest() {
        synchronized (this) {
            initialized = false;
            state = null;
            lastCommittedCounter = -1;
            registrySetupId = null;
        }
    }

}

1	package com.knowledgepixels.query;
2
3	import org.eclipse.rdf4j.common.transaction.IsolationLevels;
4	import org.eclipse.rdf4j.model.IRI;
5	import org.eclipse.rdf4j.model.Literal;
6	import org.eclipse.rdf4j.model.ValueFactory;
7	import org.eclipse.rdf4j.model.impl.SimpleValueFactory;
8	import org.eclipse.rdf4j.repository.RepositoryConnection;
9	import org.nanopub.vocabulary.NPA;
10
11	import java.util.Objects;
12
13	/**
14	* Class to control the load status of the database.
15	*/
16	public class StatusController {	6✔
17
18	/**
19	* The load states in which the database can be.
20	*/
21	public enum State {	9✔
22	/**
23	* The service is launching.
24	*/
25	LAUNCHING,	18✔
26	/**
27	* The service is loading.
28	*/
29	LOADING_INITIAL,	18✔
30	/**
31	* The service is loading updates.
32	*/
33	LOADING_UPDATES,	18✔
34	/**
35	* The service is ready to serve requests.
36	*/
37	READY,	18✔
38	/**
39	* The service detected a registry reset and is about to resync.
40	*/
41	RESETTING,	18✔
42	}
43
44	/**
45	* Get the singleton instance of the StatusController.
46	*
47	* @return the StatusController instance
48	*/
49	public static StatusController get() {
50	return instance;	6✔
51	}
52
53	private final static StatusController instance = new StatusController();	12✔
54
55	static final IRI HAS_REGISTRY_SETUP_ID =
56	SimpleValueFactory.getInstance().createIRI(NPA.NAMESPACE, "hasRegistrySetupId");	15✔
57
58	private static final ValueFactory vf = SimpleValueFactory.getInstance();	9✔
59
60	private boolean initialized = false;	9✔
61	private volatile State state = null;	9✔
62	private volatile long lastCommittedCounter = -1;	9✔
63	private volatile Long registrySetupId = null;	12✔
64
65	/**
66	* Opens a fresh admin-repo connection for a single transaction.
67	*
68	* <p>This class used to hold one connection for the lifetime of the process,
69	* assigned once in {@link #initialize()} and never re-acquired. An RDF4J restart
70	* — which the {@code rdf4j} healthcheck performs by design when its probes fail —
71	* orphaned that connection permanently: every subsequent admin-repo write threw,
72	* so {@code setLoadingUpdates} failed before {@code loadBatch} could run and the
73	* instance ingested nothing until the query container itself was restarted.
74	* RDF4J recovering on its own was not enough. Acquiring per transaction means a
75	* restart costs one failed tick instead of wedging the loader (incident
76	* 2026-07-31, query.knowledgepixels.com).
77	*
78	* <p>Cheap enough to do per call: {@code getConnection()} on an
79	* {@code HTTPRepository} is local and does no HTTP, as
80	* {@link TripleStore#getRepoConnection(String)} notes. Every other admin-repo
81	* caller in the codebase already works this way.
82	*
83	* @return a new connection the caller must close
84	*/
85	private static RepositoryConnection openAdminConnection() {
86	return TripleStore.get().getAdminRepoConnection();	9✔
87	}
88
89	/**
90	* Rolls back an active transaction, suppressing any failure to do so.
91	*
92	* <p>The transaction may no longer exist server-side (already committed, timed
93	* out, or lost to a restart), and that secondary failure must not mask the
94	* original one.
95	*
96	* @param conn the connection whose transaction should be rolled back
97	*/
98	private static void rollbackQuietly(RepositoryConnection conn) {
99	try {
100	if (conn.isActive()) {	9!
101	conn.rollback();	×
102	}
103	} catch (Exception rollbackException) {	×
104	// Deliberately ignored; the caller rethrows the original cause.
105	}	3✔
106	}	3✔
107
108	/**
109	* Represents the current status of the service, including the load counter.
110	*/
111	public static class LoadingStatus {
112
113	/**
114	* The current state of the service.
115	*/
116	public final State state;
117
118	/**
119	* The current load counter.
120	*/
121	public final long loadCounter;
122
123	private LoadingStatus(State state, long loadCounter) {	6✔
124	this.state = state;	9✔
125	this.loadCounter = loadCounter;	9✔
126	}	3✔
127
128	/**
129	* Create a new LoadingStatus instance.
130	*
131	* @param state the current state of the service
132	* @param loadCounter the current load counter
133	* @return a new LoadingStatus instance
134	*/
135	public static LoadingStatus of(State state, long loadCounter) {
136	return new LoadingStatus(state, loadCounter);	18✔
137	}
138
139	@Override
140	public boolean equals(Object o) {
141	if (o == null \|\| getClass() != o.getClass()) {	21✔
142	return false;	6✔
143	}
144	LoadingStatus that = (LoadingStatus) o;	9✔
145	return loadCounter == that.loadCounter && state == that.state;	45✔
146	}
147
148	@Override
149	public int hashCode() {
150	return Objects.hash(state, loadCounter);	45✔
151	}
152
153	}
154
155	/**
156	* Initialize the StatusController, fetching the last known state from the DB.
157	* This must be called right after service startup, before loading any nanopubs.
158	*
159	* @return the current state and the last committed counter
160	*/
161	public LoadingStatus initialize() {
162	synchronized (this) {	12✔
163	if (initialized) {	9✔
164	throw new IllegalStateException("Already initialized");	15✔
165	}
166	state = State.LAUNCHING;	9✔
167	try (RepositoryConnection conn = openAdminConnection()) {	6✔
168	try {
169	// Serializable, as the service state needs to be strictly consistent
170	conn.begin(IsolationLevels.SERIALIZABLE);	9✔
171	// Fetch the state from the DB
172	try (var statements = conn.getStatements(NPA.THIS_REPO, NPA.HAS_STATUS, null, NPA.GRAPH)) {	36✔
173	if (!statements.hasNext()) {	9!
174	conn.add(NPA.THIS_REPO, NPA.HAS_STATUS, stateAsLiteral(state), NPA.GRAPH);	45✔
175	} else {
176	var stateStatement = statements.next();	×
177	state = State.valueOf(stateStatement.getObject().stringValue());	×
178	}
179	}
180	// Fetch the load counter from the DB
181	try (var statements = conn.getStatements(NPA.THIS_REPO, NPA.HAS_REGISTRY_LOAD_COUNTER, null, NPA.GRAPH)) {	36✔
182	if (!statements.hasNext()) {	9!
183	conn.add(NPA.THIS_REPO, NPA.HAS_REGISTRY_LOAD_COUNTER, vf.createLiteral(-1L), NPA.GRAPH);	42✔
184	} else {
185	var counterStatement = statements.next();	×
186	var stringVal = counterStatement.getObject().stringValue();	×
187	lastCommittedCounter = Long.parseLong(stringVal);	×
188	}
189	}
190	// Fetch the registry setup ID from the DB
191	try (var statements = conn.getStatements(NPA.THIS_REPO, HAS_REGISTRY_SETUP_ID, null, NPA.GRAPH)) {	36✔
192	if (statements.hasNext()) {	9!
193	var setupIdStatement = statements.next();	×
194	registrySetupId = Long.parseLong(setupIdStatement.getObject().stringValue());	×
195	}
196	}
197	conn.commit();	6✔
198	} catch (Exception e) {	×
199	rollbackQuietly(conn);	×
200	throw new RuntimeException(e);	×
201	}	3✔
202	}
203	initialized = true;	9✔
204	return getState();	15✔
205	}
206	}
207
208	/**
209	* Get the current state of the service.
210	*
211	* @return the current state and the last committed counter
212	*/
213	public LoadingStatus getState() {
214	return LoadingStatus.of(state, lastCommittedCounter);	18✔
215	}
216
217	/**
218	* Transition the service to the LOADING_INITIAL state and update the load counter.
219	* This should be called in two situations:
220	* - By the main loading thread (after calling initialize()) to start loading the initial nanopubs.
221	* - By the initial nanopub loader, as it processes the initial nanopubs.
222	*
223	* @param loadCounter the new load counter
224	*/
225	public void setLoadingInitial(long loadCounter) {
226	synchronized (this) {	12✔
227	if (state != State.LAUNCHING && state != State.LOADING_INITIAL && state != State.RESETTING) {	36✔
228	throw new IllegalStateException("Cannot transition to LOADING_INITIAL, as the " + "current state is " + state);	24✔
229	}
230	if (state != State.RESETTING && lastCommittedCounter > loadCounter) {	27✔
231	throw new IllegalStateException("Cannot update the load counter from " + lastCommittedCounter + " to " + loadCounter);	24✔
232	}
233	updateState(State.LOADING_INITIAL, loadCounter);	12✔
234	}	9✔
235	}	3✔
236
237	/**
238	* Transition the service to the LOADING_UPDATES state and update the load counter.
239	* This should be called by the updates loader, when it starts processing new nanopubs, or
240	* when it finishes processing a batch of nanopubs.
241	*
242	* @param loadCounter the new load counter
243	*/
244	public void setLoadingUpdates(long loadCounter) {
245	synchronized (this) {	12✔
246	if (state != State.LAUNCHING && state != State.LOADING_UPDATES && state != State.READY) {	36✔
247	throw new IllegalStateException("Cannot transition to LOADING_UPDATES, as the " + "current state is " + state);	24✔
248	}
249	if (lastCommittedCounter > loadCounter) {	15✔
250	throw new IllegalStateException("Cannot update the load counter from " + lastCommittedCounter + " to " + loadCounter);	24✔
251	}
252	// Idempotence guard: skip the admin-repo rewrite if the state and counter
253	// are already where we'd set them. A no-op loop iteration of the updates
254	// loader otherwise re-writes the same two triples hundreds of times an
255	// hour over a long idle tail, each write growing the admin-repo LMDB via
256	// copy-on-write.
257	if (state == State.LOADING_UPDATES && lastCommittedCounter == loadCounter) {	27!
258	return;	×
259	}
260	updateState(State.LOADING_UPDATES, loadCounter);	12✔
261	}	9✔
262	}	3✔
263
264	/**
265	* Transition the service to the READY state.
266	* This should be called by the loaders, after they finish their work.
267	*/
268	public void setReady() {
269	synchronized (this) {	12✔
270	if (state != State.READY) {	12✔
271	updateState(State.READY, lastCommittedCounter);	15✔
272	}
273	}	9✔
274	}	3✔
275
276	/**
277	* Transition the service to the RESETTING state, resetting the load counter to -1.
278	* This is triggered when a registry reset is detected (setupId changed or counter decreased),
279	* or when FORCE_RESYNC is set on startup. Accepts any post-initialize source state:
280	* a reset discards prior load progress, so it's valid from LOADING_INITIAL (crashed mid-init),
281	* LOADING_UPDATES, READY, or RESETTING (retry after a crashed prior reset).
282	*/
283	public void setResetting() {
284	synchronized (this) {	12✔
285	if (state == State.LAUNCHING) {	12✔
286	throw new IllegalStateException("Cannot transition to RESETTING, as the " + "current state is " + state);	24✔
287	}
288	updateState(State.RESETTING, -1);	12✔
289	}	9✔
290	}	3✔
291
292	/**
293	* Get the persisted registry setup ID.
294	*
295	* @return the registry setup ID, or null if not yet known
296	*/
297	public Long getRegistrySetupId() {
298	// Lock-free read: field is volatile, writers still hold synchronized(this)
299	// so there are no concurrent writers. applyGlobalHeaders in MainVerticle
300	// calls this on every inbound request on the Vert.x event loop — blocking
301	// here behind updateState's admin-repo transaction was a BlockedThreadChecker
302	// hazard. The DB-commit-first order in the setter (setRegistrySetupId)
303	// means a reader can observe the previous value for the few ms between DB
304	// commit and field assignment; no caller depends on stronger consistency.
305	return registrySetupId;	9✔
306	}
307
308	/**
309	* Persist a new registry setup ID to the admin repo.
310	*
311	* @param setupId the new setup ID
312	*/
313	public void setRegistrySetupId(long setupId) {
314	synchronized (this) {	12✔
315	try (RepositoryConnection conn = openAdminConnection()) {	6✔
316	try {
317	conn.begin(IsolationLevels.SERIALIZABLE);	9✔
318	conn.remove(NPA.THIS_REPO, HAS_REGISTRY_SETUP_ID, null, NPA.GRAPH);	33✔
319	conn.add(NPA.THIS_REPO, HAS_REGISTRY_SETUP_ID, vf.createLiteral(setupId), NPA.GRAPH);	39✔
320	conn.commit();	6✔
321	registrySetupId = setupId;	12✔
322	} catch (Exception e) {	×
323	rollbackQuietly(conn);	×
324	throw new RuntimeException(e);	×
325	}	3✔
326	}
327	}	9✔
328	}	3✔
329
330	/**
331	* Persist a state transition, then adopt it in memory.
332	*
333	* <p>The in-memory fields are assigned only after the admin-repo commit succeeds,
334	* mirroring {@link #setRegistrySetupId(long)}. The earlier order (assign first,
335	* commit second) left the two permanently divergent whenever the commit threw,
336	* because nothing rolled the fields back: the loader would then read a
337	* {@code lastCommittedCounter} ahead of the persisted one, conclude from
338	* {@code lastCommittedCounter >= targetCounter} that it was caught up, and skip
339	* those nanopubs forever while still reporting READY.
340	*
341	* <p>Readers may observe the previous state for the few ms of the commit — the
342	* same window {@link #getRegistrySetupId()} already documents, and no caller
343	* depends on stronger consistency. If a commit lands server-side but the response
344	* is lost, the fields stay behind the DB instead of ahead of it; the next
345	* transition re-writes the same triples (the remove/add pair is idempotent), so
346	* that direction costs at most some re-processing rather than silent data loss.
347	*
348	* @param newState the state to transition to
349	* @param loadCounter the load counter to persist alongside it
350	*/
351	void updateState(State newState, long loadCounter) {
352	synchronized (this) {	12✔
353	try (RepositoryConnection conn = openAdminConnection()) {	6✔
354	try {
355	// Serializable, as the service state needs to be strictly consistent
356	conn.begin(IsolationLevels.SERIALIZABLE);	9✔
357	conn.remove(NPA.THIS_REPO, NPA.HAS_STATUS, null, NPA.GRAPH);	33✔
358	conn.add(NPA.THIS_REPO, NPA.HAS_STATUS, stateAsLiteral(newState), NPA.GRAPH);	39✔
359	conn.remove(NPA.THIS_REPO, NPA.HAS_REGISTRY_LOAD_COUNTER, null, NPA.GRAPH);	33✔
360	conn.add(NPA.THIS_REPO, NPA.HAS_REGISTRY_LOAD_COUNTER, vf.createLiteral(loadCounter), NPA.GRAPH);	39✔
361	conn.commit();	6✔
362	state = newState;	9✔
363	lastCommittedCounter = loadCounter;	9✔
364	} catch (Exception e) {	3✔
365	rollbackQuietly(conn);	6✔
366	throw new RuntimeException(e);	15✔
367	}	3✔
368	}
369	}	9✔
370	}	3✔
371
372	private Literal stateAsLiteral(State s) {
373	return vf.createLiteral(s.toString());	15✔
374	}
375
376	/**
377	* Reset the StatusController for testing purposes.
378	* This will clear the state and allow re-initialization.
379	*/
380	void resetForTest() {
381	synchronized (this) {	12✔
382	initialized = false;	9✔
383	state = null;	9✔
384	lastCommittedCounter = -1;	9✔
385	registrySetupId = null;	9✔
386	}	9✔
387	}	3✔
388
389	}

knowledgepixels / nanopub-query / 30629636856

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