24664456478

Committed 20 Apr 2026 11:39AM UTC coverage: 59.905% (-0.3%) from 60.236%

Build # 24664456478

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #75 from knowledgepixels/fix/timeouts-backoff-breaker

fix/perf/feat: changes 1+7+8 — HTTP timeouts, exponential backoff, circuit breaker

Coverage Stats

293 of 544 branches covered (53.86%)

Branch coverage included in aggregate %.

841 of 1349 relevant lines covered (62.34%)

6.06 hits per line

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

63.3

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

package com.knowledgepixels.query;

import org.apache.commons.exec.environment.EnvironmentUtils;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpUriRequest;
import org.apache.http.client.methods.RequestBuilder;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.BasicResponseHandler;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.eclipse.rdf4j.common.transaction.IsolationLevels;
import org.eclipse.rdf4j.model.ValueFactory;
import org.eclipse.rdf4j.model.impl.SimpleValueFactory;
import org.eclipse.rdf4j.repository.Repository;
import org.eclipse.rdf4j.repository.RepositoryConnection;
import org.eclipse.rdf4j.repository.base.RepositoryConnectionWrapper;
import org.eclipse.rdf4j.repository.http.HTTPRepository;
import org.nanopub.NanopubUtils;
import org.nanopub.vocabulary.NPA;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * Class to access the database in the form of triple stores.
 */
public class TripleStore {

    /**
     * Name of the admin graph.
     */
    public static final String ADMIN_REPO = "admin";

    private static ValueFactory vf = SimpleValueFactory.getInstance();

    private static final Logger log = LoggerFactory.getLogger(TripleStore.class);

    private final Map<String, Repository> repositories = new LinkedHashMap<>();

    /**
     * Per-repo open-connection counter, read by the eviction loop to skip repos that
     * still have live connections. Incremented in {@link #getRepoConnection(String)}
     * just before the connection is handed out and decremented via a
     * {@link RepositoryConnectionWrapper} that intercepts {@code close()} exactly once.
     */
    private final ConcurrentHashMap<String, AtomicInteger> openConnections = new ConcurrentHashMap<>();

    private String endpointBase = null;
    private String endpointType = null;

    private static TripleStore tripleStoreInstance;

    /**
     * Returns singleton triple store instance.
     *
     * @return Triple store instance
     */
    public static TripleStore get() {
        if (tripleStoreInstance == null) {
            try {
                tripleStoreInstance = new TripleStore();
            } catch (IOException ex) {
                log.info("Could not init TripleStore. ", ex);
            }
        }
        return tripleStoreInstance;
    }

    private TripleStore() throws IOException {
        Map<String, String> env = EnvironmentUtils.getProcEnvironment();
        endpointBase = env.get("ENDPOINT_BASE");
        log.info("Endpoint base: {}", endpointBase);
        endpointType = env.get("ENDPOINT_TYPE");

        getRepository("empty");  // Make sure empty repo exists
    }

    /**
     * Shared HTTP client for all RDF4J traffic. Apache HttpClient treats all requests
     * to a given host + port + protocol as one "route", so every `HTTPRepository` in
     * this process funnels through this single client's connection pool — plus the
     * two ad-hoc users in {@link #createRepo} and {@link #getRepositoryNames}, which
     * have been consolidated onto this client too.
     *
     * <p>The Apache defaults (maxPerRoute=2 / maxTotal=20) throttle four concurrent
     * loader-pool threads down to two-way parallelism at the HTTP layer — invisible
     * client-side serialisation the code is actively fighting. Raised to 10 / 40
     * here: comfortable headroom for the 4-thread loader pool plus admin-repo
     * transactions and the metrics tick, small enough to be a conservative first
     * step with room to grow later.
     *
     * <p>Timeouts set via {@code setDefaultRequestConfig}:
     * <ul>
     *   <li><b>socket-read = 60 s</b> — an individual HTTP response body must arrive
     *       within this window. Healthy per-nanopub commits complete in milliseconds,
     *       so 60 s is pure safety margin; its job is to turn the silent "threads
     *       parked forever inside a commit" wedge (observed in the April test) into
     *       a recoverable error that feeds the existing retry path.</li>
     *   <li><b>connection-request = 30 s</b> — a caller waiting for a pooled connection
     *       gives up after this long. Prevents the invisible self-deadlock in
     *       {@code loadInvalidateStatements} (one thread holding N connections while
     *       waiting for an (N+1)th) from hanging forever.</li>
     *   <li><b>connect = 10 s</b> — kills TCP handshakes that stall. Generous but
     *       bounded.</li>
     * </ul>
     * Without these defaults, HttpClient uses {@code -1} everywhere, which means
     * "wait forever".
     */
    private final CloseableHttpClient httpclient = HttpClients.custom()
            .setMaxConnPerRoute(10)
            .setMaxConnTotal(40)
            .setDefaultRequestConfig(RequestConfig.custom()
                    .setSocketTimeout(60_000)
                    .setConnectionRequestTimeout(30_000)
                    .setConnectTimeout(10_000)
                    .build())
            .build();

    @GeneratedFlagForDependentElements
    Repository getRepository(String name) {
        synchronized (this) {
            if (repositories.size() > 100) {
                evictIdleRepos();
            }
            if (repositories.containsKey(name)) {
                // Move to the end of the list:
                Repository repo = repositories.remove(name);
                repositories.put(name, repo);
            } else {
                Repository repository = null;
                if (endpointType == null || endpointType.equals("rdf4j")) {
                    HTTPRepository hr = new HTTPRepository(endpointBase + "repositories/" + name);
                    hr.setHttpClient(httpclient);
                    repository = hr;
//                        } else if (endpointType.equals("virtuoso")) {
//                                repository = new VirtuosoRepository(endpointBase + name, username, password);
                } else {
                    throw new RuntimeException("Unknown repository type: " + endpointType);
                }
                repositories.put(name, repository);
                createRepo(name);
                getRepoConnection(name).close();
            }
            return repositories.get(name);
        }
    }

    /**
     * Return the repository connection for the given repository name.
     *
     * @param name repository name
     * @return repository connection
     */
    @GeneratedFlagForDependentElements
    public RepositoryConnection getRepoConnection(String name) {
        // The increment has to happen under the same monitor that guards eviction,
        // otherwise another thread could evict this repo in the window between
        // getRepository() returning and the counter going above zero. getConnection()
        // on HTTPRepository is local (it doesn't do HTTP), so holding the lock here
        // is cheap.
        synchronized (this) {
            Repository repo = getRepository(name);
            if (repo == null) {
                return null;
            }
            AtomicInteger counter = openConnections.computeIfAbsent(name, k -> new AtomicInteger());
            counter.incrementAndGet();
            return new CountingRepositoryConnection(repo, repo.getConnection(), counter);
        }
    }

    /**
     * Evicts the eldest cache entries until either the size is back within the
     * 100-entry cap or every remaining entry has at least one open connection.
     * The cap is load-bearing — each cached entry keeps an LMDB environment alive
     * on the RDF4J server (in-memory cache, mmap pages, native memory), so
     * exceeding it by much risks server-side OOM. Actively-used repos are skipped
     * rather than shut down, because {@link org.eclipse.rdf4j.repository.http.HTTPRepository#shutDown()}
     * closes the session manager and kills any live transaction on that repo with
     * a connection-close error (the {@code MMapIndexInput – Already closed}
     * failure mode observed on query-3 in the April test). The cache self-converges
     * as active repos become idle.
     */
    @GeneratedFlagForDependentElements
    private void evictIdleRepos() {
        List<String> skipped = new ArrayList<>();
        Iterator<Entry<String, Repository>> iter = repositories.entrySet().iterator();
        while (iter.hasNext() && repositories.size() > 100) {
            Entry<String, Repository> e = iter.next();
            AtomicInteger active = openConnections.get(e.getKey());
            if (active != null && active.get() > 0) {
                skipped.add(e.getKey());
                continue;
            }
            iter.remove();
            log.info("Shutting down repo: {}", e.getKey());
            e.getValue().shutDown();
            log.info("Shutdown complete");
        }
        if (!skipped.isEmpty()) {
            log.warn("Skipped eviction for {} active repo(s); cache size is now {} (cap 100). Active names: {}",
                    skipped.size(), repositories.size(), skipped);
        }
    }

    /**
     * Minimal wrapper around a delegate {@link RepositoryConnection} whose only job
     * is to decrement the per-repo open-connection counter exactly once when the
     * caller closes it. Uses {@link AtomicBoolean} so that repeated/idempotent
     * {@code close()} calls (common with try-with-resources plus explicit close)
     * don't decrement more than once.
     */
    private static final class CountingRepositoryConnection extends RepositoryConnectionWrapper {

        private final AtomicInteger counter;
        private final AtomicBoolean closed = new AtomicBoolean();

        CountingRepositoryConnection(Repository repo, RepositoryConnection delegate, AtomicInteger counter) {
            super(repo, delegate);
            this.counter = counter;
        }

        @Override
        public void close() {
            try {
                super.close();
            } finally {
                if (closed.compareAndSet(false, true)) {
                    counter.decrementAndGet();
                }
            }
        }

    }

    @GeneratedFlagForDependentElements
    private void createRepo(String repoName) {
        if (!repoName.equals(ADMIN_REPO)) {
            getRepository(ADMIN_REPO);  // make sure admin repo is loaded first
        }
        // Uses the shared this.httpclient rather than a per-call client so it inherits
        // the configured pool sizes (and, once change 1 of the fix plan lands, the
        // socket/connection-request timeouts).
        try {
            //log.info("Trying to creating repo " + name);

            // TODO new syntax somehow doesn't work for the Lucene case:

//                        String createRegularRepoQueryString = "@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#>.\n"
//                                        + "@prefix config: <tag:rdf4j.org,2023:config/>.\n"
//                                        + "[] a config:Repository ;\n"
//                                        + "    config:rep.id \"" + name + "\" ;\n"
//                                        + "    rdfs:label \"" + name + " native store\" ;\n"
//                                        + "    config:rep.impl [\n"
//                                        + "        config:rep.type \"openrdf:SailRepository\" ;\n"
//                                        + "        config:sail.impl [\n"
//                                        + "            config:sail.type \"openrdf:NativeStore\" ;\n"
//                                        + "            config:sail.iterationCacheSyncThreshold \"10000\";\n"
//                                        + "            config:native.tripleIndexes \"spoc,posc,ospc,opsc,psoc,sopc,spoc,cpos,cosp,cops,cpso,csop\" ;\n"
//                                        + "            config:sail.defaultQueryEvaluationMode \"STANDARD\"\n"
//                                        + "        ]\n"
//                                        + "    ].";
//                        String createTextRepoQueryString = "@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#>.\n"
//                                        + "@prefix config: <tag:rdf4j.org,2023:config/>.\n"
//                                        + "[] a config:Repository ;\n"
//                                        + "    config:rep.id \"" + name + "\" ;\n"
//                                        + "    rdfs:label \"" + name + " native store\" ;\n"
//                                        + "    config:rep.impl [\n"
//                                        + "        config:rep.type \"openrdf:SailRepository\" ;\n"
//                                        + "        config:sail.impl [\n"
//                                        + "            config:sail.type \"openrdf:LuceneSail\" ;\n"
//                                        + "            config:sail.lucene.indexDir \"index/\" ;\n"
//                                        + "            config:delegate [\n"
//                                        + "                config:rep.type \"openrdf:SailRepository\" ;\n"
//                                        + "                config:sail.impl [\n"
//                                        + "                    config:sail.type \"openrdf:NativeStore\" ;\n"
//                                        + "                    config:sail.iterationCacheSyncThreshold \"10000\";\n"
//                                        + "                    config:native.tripleIndexes \"spoc,posc,ospc,opsc,psoc,sopc,spoc,cpos,cosp,cops,cpso,csop\" ;\n"
//                                        + "                    config:sail.defaultQueryEvaluationMode \"STANDARD\"\n"
//                                        + "                ]\n"
//                                        + "            ]\n"
//                                        + "        ]\n"
//                                        + "    ].";

            String indexTypes = "spoc,posc,ospc,cspo,cpos,cosp";
            if (repoName.startsWith("meta") || repoName.startsWith("text")) {
                indexTypes = "spoc,posc,ospc";
            }

            String createRegularRepoQueryString =
                    "@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#>.\n" +
                    "@prefix rep: <http://www.openrdf.org/config/repository#>.\n" +
                    "@prefix sr: <http://www.openrdf.org/config/repository/sail#>.\n" +
                    "@prefix sail: <http://www.openrdf.org/config/sail#>.\n" +
                    "@prefix sail-luc: <http://www.openrdf.org/config/sail/lucene#>.\n" +
                    "@prefix lmdb: <http://rdf4j.org/config/sail/lmdb#>.\n" +
                    "@prefix sb: <http://www.openrdf.org/config/sail/base#>.\n" +
                    "\n" +
                    "[] a rep:Repository ;\n" +
                    "    rep:repositoryID \"" + repoName + "\" ;\n" +
                    "    rdfs:label \"" + repoName + " LMDB store\" ;\n" +
                    "    rep:repositoryImpl [\n" +
                    "        rep:repositoryType \"openrdf:SailRepository\" ;\n" +
                    "        sr:sailImpl [\n" +
                    "            sail:sailType \"rdf4j:LmdbStore\" ;\n" +
                    "            sail:iterationCacheSyncThreshold \"10000\";\n" +
                    "            lmdb:tripleIndexes \"" + indexTypes + "\" ;\n" +
                    "            sb:defaultQueryEvaluationMode \"STANDARD\"\n" +
                    "        ]\n"
                    + "    ].\n";

            // TODO Index npa:hasFilterLiteral predicate too (see https://groups.google.com/g/rdf4j-users/c/epF4Af1jXGU):
            String createTextRepoQueryString =
                    "@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#>.\n" +
                    "@prefix rep: <http://www.openrdf.org/config/repository#>.\n" +
                    "@prefix sr: <http://www.openrdf.org/config/repository/sail#>.\n" +
                    "@prefix sail: <http://www.openrdf.org/config/sail#>.\n" +
                    "@prefix sail-luc: <http://www.openrdf.org/config/sail/lucene#>.\n" +
                    "@prefix lmdb: <http://rdf4j.org/config/sail/lmdb#>.\n" +
                    "@prefix sb: <http://www.openrdf.org/config/sail/base#>.\n" +
                    "\n"
                    + "[] a rep:Repository ;\n" +
                    "    rep:repositoryID \"" + repoName + "\" ;\n" +
                    "    rdfs:label \"" + repoName + " store\" ;\n" +
                    "    rep:repositoryImpl [\n" +
                    "        rep:repositoryType \"openrdf:SailRepository\" ;\n" +
                    "        sr:sailImpl [\n" +
                    "            sail:sailType \"openrdf:LuceneSail\" ;\n" +
                    "            sail-luc:indexDir \"index/\" ;\n" +
                    "            sail-luc:transactional false ;\n" +
                    "            sail:delegate [\n" +
                    "              sail:sailType \"rdf4j:LmdbStore\" ;\n" +
                    "              sail:iterationCacheSyncThreshold \"10000\";\n" +
                    "              lmdb:tripleIndexes \"" + indexTypes + "\" ;\n" +
                    "              sb:defaultQueryEvaluationMode \"STANDARD\"\n" +
                    "            ]\n" +
                    "        ]\n" +
                    "    ].";

            String createRepoQueryString = createRegularRepoQueryString;
            if (repoName.startsWith("text")) {
                createRepoQueryString = createTextRepoQueryString;
            }

            HttpUriRequest createRepoRequest = RequestBuilder.put().setUri(endpointBase + "repositories/" + repoName).addHeader("Content-Type", "text/turtle").setEntity(new StringEntity(createRepoQueryString)).build();

            // Response is try-with-resources'd so the connection is released back to
            // the shared pool rather than leaked.
            try (CloseableHttpResponse response = httpclient.execute(createRepoRequest)) {
                int statusCode = response.getStatusLine().getStatusCode();
                if (statusCode == 409) {
                    //log.info("Already exists.");
                    getRepository(repoName).init();
                } else if (statusCode >= 200 && statusCode < 300) {
                    //log.info("Successfully created.");
                    initNewRepo(repoName);
                } else {
                    log.info("Status code: {}", response.getStatusLine().getStatusCode());
                    log.info(response.getStatusLine().getReasonPhrase());
                    String handledResponse = new BasicResponseHandler().handleResponse(response);
                    log.info("Response: {}", handledResponse);
                }
            }
        } catch (IOException ex) {
            log.info("Could not create repo.", ex);
        }
    }

    /**
     * Sends shutdown signal to all repositories.
     */
    @GeneratedFlagForDependentElements
    public void shutdownRepositories() {
        for (Repository repo : repositories.values()) {
            if (repo != null && repo.isInitialized()) {
                repo.shutDown();
            }
        }
    }

    /**
     * Returns admin repo connection.
     *
     * @return repository connection to admin repository
     */
    @GeneratedFlagForDependentElements
    public RepositoryConnection getAdminRepoConnection() {
        return get().getRepoConnection(ADMIN_REPO);
    }

    private Set<String> cachedRepositoryNames = Set.of();
    private boolean repoNamesCacheValid = false;
    private final ReadWriteLock repoNamesCacheLock = new ReentrantReadWriteLock();

    /**
     * Returns set of all repository names.
     *
     * @return Repository name set
     */
    public Set<String> getRepositoryNames() {
        // See if the repository names are cached:
        final var readLock = repoNamesCacheLock.readLock();
        try {
            readLock.lock();
            if (repoNamesCacheValid) {
                return cachedRepositoryNames;
            }
        } finally {
            readLock.unlock();
        }

        // Not cached, get from server:
        final var writeLock = repoNamesCacheLock.writeLock();
        try {
            writeLock.lock();
            // Check again if another thread has already updated the cache:
            if (repoNamesCacheValid) {
                return cachedRepositoryNames;
            }
            Map<String, Boolean> repositoryNames = null;
            // Uses the shared this.httpclient; response try-with-resources releases
            // the pooled connection when done.
            try (CloseableHttpResponse resp = httpclient.execute(RequestBuilder.get()
                    .setUri(endpointBase + "/repositories")
                    .addHeader("Content-Type", "text/csv")
                    .build())) {
                BufferedReader reader = new BufferedReader(new InputStreamReader(resp.getEntity().getContent()));
                int code = resp.getStatusLine().getStatusCode();
                if (code < 200 || code >= 300) return null;
                repositoryNames = new HashMap<>();
                int lineCount = 0;
                while (true) {
                    String line = reader.readLine();
                    if (line == null) break;
                    if (lineCount > 0) {
                        String repoName = line.split(",")[1];
                        repositoryNames.put(repoName, true);
                    }
                    lineCount = lineCount + 1;
                }
            } catch (IOException ex) {
                log.info("Could not get repository names.", ex);
                return null;
            }
            cachedRepositoryNames = repositoryNames.keySet();
            repoNamesCacheValid = true;
            return cachedRepositoryNames;
        } finally {
            writeLock.unlock();
        }
    }

    /**
     * Invalidates the repository names cache. Call this method when a repository is created or deleted.
     */
    private void invalidateRepositoryNamesCache() {
        final var writeLock = repoNamesCacheLock.writeLock();
        try {
            writeLock.lock();
            repoNamesCacheValid = false;
        } finally {
            writeLock.unlock();
        }
    }

    @GeneratedFlagForDependentElements
    private void initNewRepo(String repoName) {
        String repoInitId = new Random().nextLong() + "";
        getRepository(repoName).init();
        if (!repoName.equals("empty")) {
            RepositoryConnection conn = getRepoConnection(repoName);
            try (conn) {
                // Full isolation, just in case.
                conn.begin(IsolationLevels.SERIALIZABLE);
                conn.add(NPA.THIS_REPO, NPA.HAS_REPO_INIT_ID, vf.createLiteral(repoInitId), NPA.GRAPH);
                if (tracksNanopubCountAndChecksum(repoName)) {
                    conn.add(NPA.THIS_REPO, NPA.HAS_NANOPUB_COUNT, vf.createLiteral(0L), NPA.GRAPH);
                    conn.add(NPA.THIS_REPO, NPA.HAS_NANOPUB_CHECKSUM, vf.createLiteral(NanopubUtils.INIT_CHECKSUM), NPA.GRAPH);
                }
                if (repoName.startsWith("pubkey_") || repoName.startsWith("type_")) {
                    String h = repoName.replaceFirst("^[^_]+_", "");
                    conn.add(NPA.THIS_REPO, NPA.HAS_COVERAGE_ITEM, Utils.getObjectForHash(h), NPA.GRAPH);
                    conn.add(NPA.THIS_REPO, NPA.HAS_COVERAGE_HASH, vf.createLiteral(h), NPA.GRAPH);
                    conn.add(NPA.THIS_REPO, NPA.HAS_COVERAGE_FILTER, vf.createLiteral("_" + repoName), NPA.GRAPH);
                }
                conn.commit();
            }
            // Refresh repository names cache
            invalidateRepositoryNamesCache();
        }
    }

    /**
     * Whether the given repo participates in the cumulative nanopub-count / XOR-checksum
     * tracking. Repos that don't hold raw nanopubs skip the
     * {@code npa:hasNanopubCount} and {@code npa:hasNanopubChecksum} initial triples —
     * leaving them at {@code 0} and the empty-XOR placeholder forever would just be
     * misleading. Currently excluded:
     * <ul>
     *   <li>{@code admin} — holds metadata only.</li>
     *   <li>{@code last30d} — content expires on a periodic cleanup.</li>
     *   <li>{@code trust} and {@code spaces} — hold derived state, not raw nanopubs.</li>
     * </ul>
     */
    private static boolean tracksNanopubCountAndChecksum(String repoName) {
        return !repoName.equals(ADMIN_REPO)
                && !repoName.equals("last30d")
                && !repoName.equals("trust")
                && !repoName.equals("spaces");
    }

}

knowledgepixels / nanopub-query / 24664456478

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