30627442792

Committed 31 Jul 2026 11:33AM UTC coverage: 58.54% (+0.3%) from 58.227%

Build # 30627442792

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #146 from knowledgepixels/fix/status-controller-commit-order

fix(status): persist state before adopting it in memory

Coverage Stats

622 of 1202 branches covered (51.75%)

Branch coverage included in aggregate %.

1808 of 2949 relevant lines covered (61.31%)

9.36 hits per line

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

81.92

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.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 boolean initialized = false;
    private volatile State state = null;
    private volatile long lastCommittedCounter = -1;
    private volatile Long registrySetupId = null;
    private RepositoryConnection adminRepoConn;

    /**
     * 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;
            adminRepoConn = TripleStore.get().getAdminRepoConnection();
            // Serializable, as the service state needs to be strictly consistent
            adminRepoConn.begin(IsolationLevels.SERIALIZABLE);
            // Fetch the state from the DB
            try (var statements = adminRepoConn.getStatements(NPA.THIS_REPO, NPA.HAS_STATUS, null, NPA.GRAPH)) {
                if (!statements.hasNext()) {
                    adminRepoConn.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 = adminRepoConn.getStatements(NPA.THIS_REPO, NPA.HAS_REGISTRY_LOAD_COUNTER, null, NPA.GRAPH)) {
                if (!statements.hasNext()) {
                    adminRepoConn.add(NPA.THIS_REPO, NPA.HAS_REGISTRY_LOAD_COUNTER, adminRepoConn.getValueFactory().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 = adminRepoConn.getStatements(NPA.THIS_REPO, HAS_REGISTRY_SETUP_ID, null, NPA.GRAPH)) {
                if (statements.hasNext()) {
                    var setupIdStatement = statements.next();
                    registrySetupId = Long.parseLong(setupIdStatement.getObject().stringValue());
                }
                adminRepoConn.commit();
            } catch (Exception e) {
                if (adminRepoConn.isActive()) {
                    try {
                        adminRepoConn.rollback();
                    } catch (Exception rollbackException) {
                        // Transaction may not be registered on server (e.g., already committed, timed out, or connection reset)
                        // Log the rollback failure but don't mask the original exception
                    }
                }
                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 {
                adminRepoConn.begin(IsolationLevels.SERIALIZABLE);
                adminRepoConn.remove(NPA.THIS_REPO, HAS_REGISTRY_SETUP_ID, null, NPA.GRAPH);
                adminRepoConn.add(NPA.THIS_REPO, HAS_REGISTRY_SETUP_ID, adminRepoConn.getValueFactory().createLiteral(setupId), NPA.GRAPH);
                adminRepoConn.commit();
                registrySetupId = setupId;
            } catch (Exception e) {
                if (adminRepoConn.isActive()) {
                    try {
                        adminRepoConn.rollback();
                    } catch (Exception rollbackException) {
                        // Log the rollback failure but don't mask the original exception
                    }
                }
                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 {
                // Serializable, as the service state needs to be strictly consistent
                adminRepoConn.begin(IsolationLevels.SERIALIZABLE);
                adminRepoConn.remove(NPA.THIS_REPO, NPA.HAS_STATUS, null, NPA.GRAPH);
                adminRepoConn.add(NPA.THIS_REPO, NPA.HAS_STATUS, stateAsLiteral(newState), NPA.GRAPH);
                adminRepoConn.remove(NPA.THIS_REPO, NPA.HAS_REGISTRY_LOAD_COUNTER, null, NPA.GRAPH);
                adminRepoConn.add(NPA.THIS_REPO, NPA.HAS_REGISTRY_LOAD_COUNTER, adminRepoConn.getValueFactory().createLiteral(loadCounter), NPA.GRAPH);
                adminRepoConn.commit();
                state = newState;
                lastCommittedCounter = loadCounter;
            } catch (Exception e) {
                if (adminRepoConn.isActive()) {
                    try {
                        adminRepoConn.rollback();
                    } catch (Exception rollbackException) {
                        // Transaction may not be registered on server (e.g., already committed, timed out, or connection reset)
                        // Log the rollback failure but don't mask the original exception
                    }
                }
                throw new RuntimeException(e);
            }
        }
    }

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

knowledgepixels / nanopub-query / 30627442792

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