diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..4b558d3 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,130 @@ +# Changelog + +## Unreleased + +### Changed + +- **`triples_update` now performs a general atomic update, not only equal-length + replacement** (`graph_db_interface/queries/triple_multi.py`). + + Previously `triples_update` raised `InvalidInputError` unless `old_triples` and + `new_triples` had the same length, restricting it to 1:1 value replacement. The + method already builds a single `DELETE ... INSERT ... WHERE` SPARQL transaction, + which handles pure additions, pure removals, and unequal-size replacements + equally well, so the length restriction was removed. The existence pre-check is + now skipped when `old_triples` is empty (a pure insert has nothing to check). + + This atomicity is required for constraint (SHACL) correctness: replacing a + cardinality-constrained property — e.g. a possession handover under a "possessed + by exactly one resource" shape — must apply the removal and the insertion in one + transaction so the intermediate (property-absent) state is never validated. + + Enables `kapps_ogm.OGM.commit` to add/remove/replace properties through a single + atomic transaction. Requested by the `kapps_semantic_middleware` project. + +### Fixed + +- **Every request opened a new TCP connection and re-ran the TLS handshake** + (`graph_db_interface/graph_db.py`). + + **Symptom:** every call through the client — query, update, graph import, repository + listing — cost far more than the work it asked the server to do. Against a remote HTTPS + endpoint (`https://graphdb.iam-mms.kit.edu`) a trivial `ASK { ?s ?p ?o }` measured + ~17 ms, of which only ~4 ms was the request itself. The cost was invisible per call and + compounded with volume: a single `kapps_semantic_middleware` integration test issues 129 + requests, so it paid ~1.7 s in handshakes alone. + + **Cause:** `_make_request` dispatched through the module-level `requests` helpers, + `getattr(requests, method)(...)`. Those are documented convenience wrappers that + construct a `requests.Session`, use it for exactly one request, and close it. Closing the + session discards its `urllib3` connection pool, so no connection was ever reused and each + call paid a fresh TCP connect plus a full TLS handshake. Confirmed by counting + `urllib3` connection creations: three sequential queries opened three + `HTTPSConnectionPool` connections. + + **Fix:** the client now holds a persistent `requests.Session` **per thread** and issues + every request through it, so the connection pool survives across calls. Per thread rather + than one shared session because a `requests.Session` mutates its cookie jar on every + response and is not thread-safe — and one client is commonly driven from several threads + at once, e.g. a web framework serving requests while the embedding code queries. urllib3's + connection pools are thread-safe; the session wrapping them is not. Each thread pays one + handshake and then reuses its own pool. Added `GraphDB.close()`, which releases the pools + held for every thread that used the client. No change to any method signature or response + handling. + + **Measured effect:** per-request cost against the remote endpoint dropped from ~17 ms to + ~4 ms. The `kapps_semantic_middleware` suite (146 tests, live GraphDB) went from 158 s to + 60 s — a 2.65× speed-up with no test changes. Regression test: + `tests/test_connection_reuse.py`. + +- **`triples_update` replaced a retained blank node instead of updating it** + (`graph_db_interface/queries/triple_multi.py`). + + **Symptom:** updating one property of an existing blank node was impossible. The node + was unlinked and a *different* node took its place, so every triple attached to the + original that the caller had not listed in `old_triples` became unreachable — silently, + with the call reporting success. + + **Cause:** the DELETE and INSERT patterns were rendered from two separate blank-node → + variable maps (`old_bn_var_map` / `new_bn_var_map`). A `BNode` passed on both sides — + the caller's way of saying "same node, different property value" — therefore became + `?oldbn1` in the DELETE and `?newbn1` in the INSERT, and every new-side variable was + bound by `BIND(BNODE() AS ?newbnN)`, which mints a fresh store node. Captured query + before the fix, for a node whose only change is its value: + + ```sparql + DELETE { :hasValue ?oldbn1 . ?oldbn1 :hasValue 12.1 } + INSERT { :hasValue ?newbn1 . ?newbn1 :hasValue 1.4 } + WHERE { :hasValue ?oldbn1 . ?oldbn1 :hasValue 12.1 . BIND(BNODE() AS ?newbn1) } + ``` + + **Fix:** one shared map across both pattern sets, so a blank node present on both sides + renders as a single variable already bound by the WHERE clause. `BIND(BNODE() AS ?v)` is + now emitted only for blank nodes exclusive to `new_triples`. Behaviour for pure + additions, pure removals, unequal-length replacements and the IRI-only path is unchanged, + as is the single-transaction atomicity the SHACL note above depends on. + + **Tests:** `tests/test_triple_update_query.py` asserts on the generated SPARQL and needs + no live repository, unlike the existing update tests. Verified red before the change and + green after: the two blank-node-identity tests fail on the previous implementation, the + two control tests pass on both. + + Found while designing anonymous-node identity in `kapps_ogm` (see + `JaFeKl/graph_db_interface#6`); a correctness fix for this library independently of that + work. Reported by the `kapps_semantic_middleware` project. + +- **`IRI` rejected any URL containing a port** (`graph_db_interface/utils/iri.py`, + `IRI._sanitize`). + + **Symptom:** constructing an `IRI` from a perfectly valid `http`/`https` URL that + includes a port raised `InvalidIRIError: ':' outside of supported schemes`: + + ```python + IRI("http://127.0.0.1:8991") # raised + IRI("http://host:8991/workflows/x/execute") # raised + ``` + + This also broke reads: `triples_get` / `query(convert_bindings=True)` convert an + `xsd:anyURI` literal to an `IRI` via `from_xsd_literal`, so any stored + `xsd:anyURI` value that was a ported URL made the read itself throw. + + **Root cause:** for a full IRI beginning with a known scheme, the validator + rejected the whole string when `raw.count(":") > 1`. A scheme contributes one + colon (`http://`) and an authority port contributes a second (`:8991`), so every + ported URL tripped the check. The check's real intent is to catch a `:` used + where a `#` was meant (e.g. `http://…/owl:Class`). + + **Fix:** after confirming the scheme, strip the scheme, split off the authority + (up to the first `/`, `#`, or `?`), remove a trailing `:` port from the + authority, and only then reject if a stray `:` remains in the authority or the + remainder. Ports are now accepted; the previously-rejected malformed forms + (`http://www.w3.org/2002/07:owl#`, `http://www.w3.org/2002/07/owl:Class`, etc.) + are still rejected. All existing `tests/test_iri.py` cases continue to pass. + + **Reported/fixed by:** the `kapps_semantic_middleware` project, whose Service + ontology stores middleware endpoint URLs (`svc:address`, `svc:endpoint`) — always + `host:port` values — as `xsd:anyURI`, and which hit this on the first round-trip + of a registered endpoint. Per that project's dependency-and-bugfix policy + (`kapps_semantic_middleware/docs/adr/0001-dependency-wiring-and-bugfix-policy.md`), + genuine correctness bugs in sibling dependency repos are fixed directly in the + sibling with a detailed changelog entry — this is that entry. diff --git a/graph_db_interface/graph_db.py b/graph_db_interface/graph_db.py index aefd197..895d41c 100644 --- a/graph_db_interface/graph_db.py +++ b/graph_db_interface/graph_db.py @@ -5,6 +5,7 @@ from graph_db_interface.kafka.kafka_manager import KafkaManager import requests import logging +import threading from requests import Response from graph_db_interface.utils.graph_db_credentials import GraphDBCredentials from graph_db_interface.utils.iri import IRI @@ -44,6 +45,21 @@ def __init__( self._auth = None self._blank_ids = set() + # A persistent session per thread, so the connection pool survives across calls. + # The module-level `requests.get`/`requests.post` helpers build a throwaway session + # per call, which means a new TCP connect and a full TLS handshake for every SPARQL + # query — against a remote HTTPS endpoint that dominates the cost of a query, and a + # client typically issues many. + # + # Per thread rather than one shared session, because a `requests.Session` is not + # thread-safe: its cookie jar is mutated by every response. One client is commonly + # driven from several threads at once — a web framework serving requests while the + # embedding code queries — and urllib3's pools are thread-safe but the session + # around them is not. Each thread pays one handshake and then reuses its own pool. + self._thread_local = threading.local() + self._sessions: list[requests.Session] = [] + self._sessions_lock = threading.Lock() + if use_gdb_token: self._auth = self._get_authentication_token( self._credentials.username, self._credentials.password @@ -243,13 +259,32 @@ def _make_request( if self._auth is not None: headers["Authorization"] = self._auth - return getattr(requests, method)( + return self._session.request( + method, f"{self._credentials.base_url}/{endpoint}", headers=headers, timeout=timeout, **kwargs, ) + @property + def _session(self) -> requests.Session: + """This thread's persistent session, created on first use.""" + session = getattr(self._thread_local, "session", None) + if session is None: + session = requests.Session() + self._thread_local.session = session + with self._sessions_lock: + self._sessions.append(session) + return session + + def close(self) -> None: + """Release the HTTP connection pools held for every thread that used this client.""" + with self._sessions_lock: + sessions, self._sessions = self._sessions, [] + for session in sessions: + session.close() + def _get_authentication_token( self, username: str, diff --git a/graph_db_interface/queries/triple_multi.py b/graph_db_interface/queries/triple_multi.py index 6879b7d..aa47c94 100644 --- a/graph_db_interface/queries/triple_multi.py +++ b/graph_db_interface/queries/triple_multi.py @@ -333,9 +333,17 @@ def triples_update( """ Update multiple RDF triples in the triplestore. + The removal of `old_triples` and the insertion of `new_triples` are applied in a + single atomic `DELETE ... INSERT ... WHERE` SPARQL transaction. The two lists need + not be the same length: this supports pure additions, pure removals, and general + replacements. Atomicity matters for constraint (e.g. SHACL) correctness — replacing + a cardinality-constrained property (such as a possession handover under a + "possessed by exactly one resource" shape) must never expose the intermediate state + in which the property is momentarily absent. + Args: - old_triples (TriplesLike): Triples to be replaced. - new_triples (TriplesLike): Replacement triples (same length as `old_triples`). + old_triples (TriplesLike): Triples to remove. + new_triples (TriplesLike): Triples to insert. Need not match the length of `old_triples`. check_exist (Optional[bool]): If True, abort when any old triple does not exist. Defaults to True. named_graph (Optional[GraphNameLike]): Override the client's default named graph. @@ -347,14 +355,15 @@ def triples_update( if not old_triples and not new_triples: return True - if len(old_triples) != len(new_triples): - raise InvalidInputError("Old and new triples lists must have the same length.") - validated_old_triples = [utils.sanitize_triple(triple) for triple in old_triples] validated_new_triples = [utils.sanitize_triple(triple) for triple in new_triples] - if check_exist and not self.all_triple_exists( - triples=validated_old_triples, named_graph=named_graph + if ( + check_exist + and validated_old_triples + and not self.all_triple_exists( + triples=validated_old_triples, named_graph=named_graph + ) ): self.logger.warning( "At least one of the triples to update does not exist in the graph." @@ -391,13 +400,23 @@ def _build_patterns( patterns.append(f"{subj_str} {pred_str} {obj_str} .") return patterns - old_bn_var_map: Dict[BNode, str] = {} - new_bn_var_map: Dict[BNode, str] = {} - - old_delete_patterns = _build_patterns( - validated_old_triples, old_bn_var_map, "oldbn" - ) - insert_patterns = _build_patterns(validated_new_triples, new_bn_var_map, "newbn") + # One map across both sides: a BNode occurring in old_triples and in new_triples is + # the *same* node, so it must render as the same SPARQL variable. Two maps gave it + # two variables, and the new-side one was then bound by BIND(BNODE()) below — which + # minted a replacement node and orphaned every triple the caller had not listed in + # old_triples. That made it impossible to update one property of an existing blank + # node while keeping the node. + bn_var_map: Dict[BNode, str] = {} + + old_delete_patterns = _build_patterns(validated_old_triples, bn_var_map, "bn") + bn_bound_by_where = set(bn_var_map) + insert_patterns = _build_patterns(validated_new_triples, bn_var_map, "bn") + + # Only blank nodes exclusive to the new side need minting; the rest are already + # bound by the WHERE clause. + new_only_bn_vars = [ + var for bnode, var in bn_var_map.items() if bnode not in bn_bound_by_where + ] def _format_block(patterns: List[str]) -> str: if not patterns: @@ -411,9 +430,9 @@ def _format_block(patterns: List[str]) -> str: where_block_parts: List[str] = [] if where_patterns: where_block_parts.append(_format_block(where_patterns)) - if new_bn_var_map: + if new_only_bn_vars: where_block_parts.extend( - f" BIND(BNODE() AS {var})" for var in new_bn_var_map.values() + f" BIND(BNODE() AS {var})" for var in new_only_bn_vars ) where_block = "\n".join(where_block_parts) diff --git a/graph_db_interface/utils/iri.py b/graph_db_interface/utils/iri.py index a2e7b53..f25f1a9 100644 --- a/graph_db_interface/utils/iri.py +++ b/graph_db_interface/utils/iri.py @@ -304,7 +304,24 @@ def _sanitize( # Full IRI if any(raw.startswith(scheme) for scheme in IRI.SCHEMES): - if raw.count(":") > 1: + # Reject a ':' used where a '#' was intended (e.g. ".../owl:Class"), + # but allow a legitimate authority port (e.g. "http://host:8991/path"). + # The scheme's own ':' (in "://") is stripped first; then a trailing + # ':' port in the authority is removed before counting any + # remaining stray colons. + matched_scheme = next(s for s in IRI.SCHEMES if raw.startswith(s)) + after_scheme = raw[len(matched_scheme):] + sep_index = len(after_scheme) + for sep in ("/", "#", "?"): + i = after_scheme.find(sep) + if i != -1: + sep_index = min(sep_index, i) + authority = after_scheme[:sep_index] + remainder = after_scheme[sep_index:] + host_part, sep, port_part = authority.rpartition(":") + if sep and port_part.isdigit(): + authority = host_part + if authority.count(":") + remainder.count(":") > 0: raise InvalidIRIError( f"Invalid IRI: ':' outside of supported schemes {IRI.SCHEMES} ({value}, {base})" ) diff --git a/tests/test_connection_reuse.py b/tests/test_connection_reuse.py new file mode 100644 index 0000000..c2ceaf0 --- /dev/null +++ b/tests/test_connection_reuse.py @@ -0,0 +1,77 @@ +import threading + +import urllib3.connectionpool +import pytest + +from graph_db_interface import GraphDB + + +@pytest.fixture +def count_new_connections(monkeypatch): + """Count TCP connections opened by urllib3 while the fixture is active. + + A client that reuses one session keeps urllib3's connection pool alive between + calls, so repeated queries open no new connection. A client that goes through the + module-level `requests` helpers builds a fresh session — and therefore a fresh + pool — per call, so every query opens one. + """ + opened = [] + + for pool_cls in ( + urllib3.connectionpool.HTTPConnectionPool, + urllib3.connectionpool.HTTPSConnectionPool, + ): + original = pool_cls._new_conn + + def counting_new_conn(self, _original=original): + opened.append(type(self).__name__) + return _original(self) + + monkeypatch.setattr(pool_cls, "_new_conn", counting_new_conn) + + return opened + + +def test_repeated_queries_reuse_one_connection(db: GraphDB, count_new_connections): + """Repeated queries must not re-handshake. + + Against a remote HTTPS endpoint the TCP connect plus TLS handshake costs more than + the query itself, so a per-call connection makes every query several times slower. + """ + for _ in range(3): + db.query("ASK { ?s ?p ?o }") + + assert count_new_connections == [], ( + f"expected the warm connection pool to be reused, but " + f"{len(count_new_connections)} new connection(s) were opened: {count_new_connections}" + ) + + +def test_client_holds_a_persistent_session(db: GraphDB): + """The session is one object for the thread's lifetime, not one per request.""" + before = db._session + db.query("ASK { ?s ?p ?o }") + + assert db._session is before + + +def test_each_thread_gets_its_own_session(db: GraphDB): + """One client is driven from several threads at once — a web framework serving requests + while the embedding code queries. A `requests.Session` mutates its cookie jar on every + response and is not thread-safe, so threads must not share one. + """ + sessions = [] + + def query_from_thread(): + db.query("ASK { ?s ?p ?o }") + sessions.append(db._session) + + threads = [threading.Thread(target=query_from_thread) for _ in range(3)] + for thread in threads: + thread.start() + for thread in threads: + thread.join(timeout=30) + + assert len(sessions) == 3, "a worker thread failed to complete its query" + assert len(set(map(id, sessions))) == 3, "threads shared a session" + assert db._session not in sessions, "a worker thread reused the main thread's session" diff --git a/tests/test_graph_manipulation.py b/tests/test_graph_manipulation.py index 3386a26..3842ea3 100644 --- a/tests/test_graph_manipulation.py +++ b/tests/test_graph_manipulation.py @@ -301,6 +301,48 @@ def test_update_multiple_triples(db: GraphDB, named_graph: str): assert result is True +ASYM_SUB = "http://example.org#asymmetric_subject" +ASYM_PRED = "http://example.org#asymmetric_predicate" + + +def test_update_adds_more_triples_than_it_removes(db: GraphDB, named_graph: str): + """An update is not restricted to 1:1 replacement. + + `kapps_ogm.OGM.commit` feeds the output of a diff straight to `triples_update`, and a + diff is a set difference — the removed and added lists are almost never the same + length. An equal-length constraint here therefore breaks every such commit, so this + pins the widened contract rather than leaving it to a downstream consumer to discover. + """ + old = [(ASYM_SUB, ASYM_PRED, 1)] + new = [(ASYM_SUB, ASYM_PRED, 2), (ASYM_SUB, ASYM_PRED, 3), (ASYM_SUB, ASYM_PRED, 4)] + + assert db.triples_add(old, named_graph=named_graph) is True + + assert db.triples_update(old_triples=old, new_triples=new, named_graph=named_graph) is True + + assert db.triple_exists(old[0], named_graph=named_graph) is False + for triple in new: + assert db.triple_exists(triple, named_graph=named_graph) is True + + assert db.triples_delete(new, named_graph=named_graph) is True + + +def test_update_removes_more_triples_than_it_adds(db: GraphDB, named_graph: str): + """The reverse direction of the same widening — a diff that drops more than it adds.""" + old = [(ASYM_SUB, ASYM_PRED, 5), (ASYM_SUB, ASYM_PRED, 6), (ASYM_SUB, ASYM_PRED, 7)] + new = [(ASYM_SUB, ASYM_PRED, 8)] + + assert db.triples_add(old, named_graph=named_graph) is True + + assert db.triples_update(old_triples=old, new_triples=new, named_graph=named_graph) is True + + for triple in old: + assert db.triple_exists(triple, named_graph=named_graph) is False + assert db.triple_exists(new[0], named_graph=named_graph) is True + + assert db.triples_delete(new, named_graph=named_graph) is True + + def test_iri_exists(db: GraphDB, named_graph: str): # add a new triple to the default graph result = db.triple_add( diff --git a/tests/test_triple_update_query.py b/tests/test_triple_update_query.py new file mode 100644 index 0000000..9efd7c6 --- /dev/null +++ b/tests/test_triple_update_query.py @@ -0,0 +1,122 @@ +"""Query-construction tests for `triples_update`. + +These do not need a live GraphDB: they stub `query` and assert on the SPARQL text. +The blank-node behaviour they pin cannot be observed through the return value, and the +existing update tests in `test_graph_manipulation.py` all require a repository. +""" + +import logging +import re + +import pytest +from rdflib import BNode, Literal + +from graph_db_interface import GraphDB, IRI + + +SUB = IRI("https://example.org/Belt1") +PRED = IRI("https://example.org/hasSpeed") +VALUE = IRI("https://example.org/hasValue") + + +@pytest.fixture +def stub_db() -> GraphDB: + """A GraphDB that records the last query instead of sending it. + + Built with `__new__` because `__init__` authenticates against a live server. + """ + db = GraphDB.__new__(GraphDB) + db.logger = logging.getLogger("stub") + db.named_graph = None + db._repository = "stub" + db._blank_ids = set() + db.queries = [] + db.query = lambda query, update=False, **kwargs: (db.queries.append(query), True)[1] + db.all_triple_exists = lambda triples, named_graph=None: True + return db + + +def _variables(block: str) -> list[str]: + return re.findall(r"\?\w+", block) + + +def _split_blocks(query: str) -> dict[str, str]: + match = re.search( + r"DELETE \{(?P.*?)\}\s*INSERT \{(?P.*?)\}\s*WHERE \{(?P.*?)\}\s*$", + query, + re.DOTALL, + ) + assert match is not None, f"unexpected query shape:\n{query}" + return match.groupdict() + + +def test_blank_node_on_both_sides_is_one_variable(stub_db: GraphDB): + """A retained blank node must keep its identity across the update. + + Regression: the DELETE and INSERT patterns used to be rendered from two separate + blank-node maps, so the same node got two variables and the new one was minted by + BIND(BNODE()) — replacing the node and orphaning anything not listed in old_triples. + """ + node = BNode("genid-retained") + + stub_db.triples_update( + old_triples=[(SUB, PRED, node), (node, VALUE, Literal(12.1))], + new_triples=[(SUB, PRED, node), (node, VALUE, Literal(1.4))], + ) + blocks = _split_blocks(stub_db.queries[-1]) + + delete_vars = set(_variables(blocks["delete"])) + insert_vars = set(_variables(blocks["insert"])) + assert len(delete_vars) == 1 + assert delete_vars == insert_vars, "the retained node must be the same variable" + assert "BIND(BNODE()" not in blocks["where"], "a retained node must not be minted" + + +def test_blank_node_only_in_new_triples_is_minted(stub_db: GraphDB): + """A genuinely new anonymous node still gets a fresh store node.""" + fresh = BNode("genid-fresh") + + stub_db.triples_update( + old_triples=[(SUB, PRED, Literal("gone"))], + new_triples=[(SUB, PRED, fresh), (fresh, VALUE, Literal(1.4))], + ) + blocks = _split_blocks(stub_db.queries[-1]) + + insert_vars = set(_variables(blocks["insert"])) + assert len(insert_vars) == 1 + assert blocks["where"].count("BIND(BNODE()") == 1 + assert insert_vars.pop() in blocks["where"] + + +def test_mixed_retained_and_new_blank_nodes(stub_db: GraphDB): + """Only the new-side node is minted; the retained one stays bound by the WHERE.""" + retained, fresh = BNode("genid-retained"), BNode("genid-fresh") + + stub_db.triples_update( + old_triples=[(SUB, PRED, retained), (retained, VALUE, Literal(12.1))], + new_triples=[ + (SUB, PRED, retained), + (retained, VALUE, Literal(1.4)), + (SUB, PRED, fresh), + (fresh, VALUE, Literal(9.9)), + ], + ) + blocks = _split_blocks(stub_db.queries[-1]) + + assert blocks["where"].count("BIND(BNODE()") == 1 + retained_var = set(_variables(blocks["delete"])) + assert len(retained_var) == 1 + assert retained_var <= set(_variables(blocks["insert"])) + assert len(set(_variables(blocks["insert"]))) == 2 + + +def test_no_blank_nodes_is_unchanged(stub_db: GraphDB): + """The ordinary IRI-only path emits no variables and no BIND.""" + stub_db.triples_update( + old_triples=[(SUB, VALUE, Literal(12.1))], + new_triples=[(SUB, VALUE, Literal(1.4))], + ) + query = stub_db.queries[-1] + + assert "?" not in query + assert "BIND(BNODE()" not in query