Never let a class-connectivity backend failure read as count 0

src/test/test_class_connectivity_no_silent_zero.py

@@ -0,0 +1,200 @@
+"""A class-connectivity backend failure must never read as ``count: 0``.
+
+Backend-free: every Neo4j/Solr/Owlery touch point inside
+``_aggregate_class_connectivity`` is monkeypatched, so these run in CI without
+credentials and exercise exactly the paths that used to ``return []``.
+
+Why: on 2026-09-04 gamma Kenyon cell (FBbt_00100247) showed "There is no data
+to display" for DownstreamClassConnectivity. The origin had returned
+``{'count': 0}`` during a moment it could not fetch per-instance connectivity,
+the Solr result cache stored it (``count >= 0`` is "valid"), and the
+v3-cached nginx edge pinned that 200 for a month. Three caches, one lie.
+"""
+import types
+
+import pytest
+
+from vfbquery import ha_api
+from vfbquery import solr_result_cache as src
+from vfbquery import vfb_queries as vq
+
+
+CLASS = "FBbt_00100247"
+INSTANCES = {"VFB_1", "VFB_2", "VFB_3"}
+
+
+class _NC:
+    """Stand-in for ``vc.nc`` whose ``commit_list`` returns whatever we say."""
+
+    def __init__(self, reply):
+        self._reply = reply
+
+    def commit_list(self, statements):
+        reply = self._reply
+        if isinstance(reply, Exception):
+            raise reply
+        return reply
+
+
+def _membership_reply(instances=INSTANCES):
+    """A Neo4j transaction result listing *instances* under CLASS."""
+    return [{
+        "columns": ["cid", "label", "iids"],
+        "data": [{"row": [CLASS, "gamma Kenyon cell", sorted(instances)]}],
+    }]
+
+
+def _wire(monkeypatch, *, membership=None, edges=None, missing=(),
+          partner_entries=None, ancestors=None, partner_membership=None):
+    """Patch every backend seam of ``_aggregate_class_connectivity``.
+
+    Defaults describe a healthy query with one downstream partner class.
+    """
+    if membership is None:
+        membership = _membership_reply()
+    if edges is None:
+        edges = {i: [{"id": "VFB_p", "outputs": 5}] for i in INSTANCES}
+    if partner_entries is None:
+        partner_entries = [{"object": {"short_form": "FBbt_p"}}]
+    if ancestors is None:
+        ancestors = ({"FBbt_p"}, {"FBbt_p": "partner"})
+    if partner_membership is None:
+        partner_membership = {"VFB_p": {"FBbt_p"}}
+
+    fake_vc = types.SimpleNamespace(
+        nc=_NC(membership),
+        vfb=types.SimpleNamespace(oc=types.SimpleNamespace(
+            get_subclasses=lambda **kw: [])),
+    )
+    monkeypatch.setattr(vq, "vc", fake_vc)
+    monkeypatch.setattr(vq, "_bulk_fetch_per_instance_connectivity",
+                        lambda ids: (dict(edges), list(missing)))
+    monkeypatch.setattr(vq, "_fetch_connectivity_entries",
+                        lambda *a, **k: list(partner_entries))
+    monkeypatch.setattr(vq, "_get_partner_class_ancestors",
+                        lambda *a, **k: ancestors)
+    monkeypatch.setattr(vq, "_build_partner_instance_class_membership",
+                        lambda ids: dict(partner_membership))
+
+
+# ---------------------------------------------------------------------------
+# _aggregate_class_connectivity
+# ---------------------------------------------------------------------------
+
+def test_healthy_query_returns_rows(monkeypatch):
+    _wire(monkeypatch)
+    status = {}
+    rows = vq._aggregate_class_connectivity(CLASS, "downstream", status=status)
+    assert [r["id"] for r in rows] == ["FBbt_p"]
+    assert rows[0]["connected_n"] == 3 and rows[0]["total_n"] == 3
+    assert status == {"missing": 0, "total": 3}
+
+
+def test_class_with_no_instances_is_a_true_zero(monkeypatch):
+    _wire(monkeypatch, membership=[{"columns": ["cid", "label", "iids"],
+                                    "data": []}])
+    assert vq._aggregate_class_connectivity(CLASS, "downstream") == []
+
+
+def test_instances_with_no_positive_edges_is_a_true_zero(monkeypatch):
+    _wire(monkeypatch, edges={i: [] for i in INSTANCES},
+          partner_entries=[], ancestors=(set(), {}))
+    assert vq._aggregate_class_connectivity(CLASS, "downstream") == []
+
+
+def test_membership_query_exception_raises(monkeypatch):
+    _wire(monkeypatch, membership=RuntimeError("neo4j down"))
+    with pytest.raises(vq.ConnectivityBackendError):
+        vq._aggregate_class_connectivity(CLASS, "downstream")
+
+
+def test_membership_query_false_reply_raises(monkeypatch):
+    # commit_list signals a transaction error by returning False, which
+    # dict_cursor used to swallow into [] — the silent path.
+    _wire(monkeypatch, membership=False)
+    with pytest.raises(vq.ConnectivityBackendError):
+        vq._aggregate_class_connectivity(CLASS, "downstream")
+
+
+def test_no_per_instance_connectivity_at_all_raises(monkeypatch):
+    _wire(monkeypatch, edges={}, missing=sorted(INSTANCES))
+    with pytest.raises(vq.ConnectivityBackendError):
+        vq._aggregate_class_connectivity(CLASS, "downstream")
+
+
+def test_partner_classes_unresolved_despite_edges_raises(monkeypatch):
+    _wire(monkeypatch, partner_entries=[], ancestors=(set(), {}))
+    with pytest.raises(vq.ConnectivityBackendError):
+        vq._aggregate_class_connectivity(CLASS, "downstream")
+
+
+def test_partner_membership_failure_raises(monkeypatch):
+    _wire(monkeypatch, partner_membership={})
+    with pytest.raises(vq.ConnectivityBackendError):
+        vq._aggregate_class_connectivity(CLASS, "downstream")
+
+
+def test_partial_coverage_is_reported_in_status(monkeypatch):
+    edges = {"VFB_1": [{"id": "VFB_p", "outputs": 5}]}
+    _wire(monkeypatch, edges=edges, missing=["VFB_2", "VFB_3"])
+    status = {}
+    rows = vq._aggregate_class_connectivity(CLASS, "downstream", status=status)
+    assert rows and status == {"missing": 2, "total": 3}
+
+
+# ---------------------------------------------------------------------------
+# the public functions: partial flag, and the Solr cache refusing it
+# ---------------------------------------------------------------------------
+
+def test_partial_result_is_flagged_and_complete_result_is_not(monkeypatch):
+    _wire(monkeypatch)
+    full = vq.get_downstream_class_connectivity.__wrapped__(
+        CLASS, return_dataframe=False)
+    assert full["count"] == 1 and vq.PARTIAL_RESULT_KEY not in full
+
+    _wire(monkeypatch, edges={"VFB_1": [{"id": "VFB_p", "outputs": 5}]},
+          missing=["VFB_2", "VFB_3"])
+    partial = vq.get_downstream_class_connectivity.__wrapped__(
+        CLASS, return_dataframe=False)
+    assert partial["count"] == 1
+    assert partial[vq.PARTIAL_RESULT_KEY]["missing_instances"] == 2
+    assert partial[vq.PARTIAL_RESULT_KEY]["total_instances"] == 3
+    assert src.result_is_partial(partial) and not src.result_is_partial(full)
+
+
+def test_upstream_shares_the_same_paths(monkeypatch):
+    _wire(monkeypatch, membership=False)
+    with pytest.raises(vq.ConnectivityBackendError):
+        vq.get_upstream_class_connectivity.__wrapped__(
+            CLASS, return_dataframe=False)
+
+
+# ---------------------------------------------------------------------------
+# ha_api: what the edge is told
+# ---------------------------------------------------------------------------
+
+def test_empty_result_is_not_cached_at_the_edge():
+    headers = ha_api._edge_cache_headers({"headers": {}, "rows": [], "count": 0})
+    assert headers["X-Accel-Expires"] == "0"
+    assert headers["Cache-Control"] == "no-store"
+
+
+def test_partial_result_gets_a_short_edge_ttl():
+    result = {"headers": {}, "rows": [{"id": "x"}], "count": 1,
+              ha_api.PARTIAL_RESULT_KEY: {"missing_instances": 2}}
+    headers = ha_api._edge_cache_headers(result)
+    assert headers["X-Accel-Expires"] == str(ha_api.PARTIAL_RESULT_EDGE_TTL)
+    assert headers["Cache-Control"] == "max-age=%d" % ha_api.PARTIAL_RESULT_EDGE_TTL
+
+
+def test_complete_result_gets_the_edge_default():
+    assert ha_api._edge_cache_headers(
+        {"headers": {}, "rows": [{"id": "x"}], "count": 1}) == {}
+
+
+def test_page_past_the_end_is_not_empty():
+    # count is the authority: a later page of a 40-row result has no rows
+    # in this slice but is not a "no data" answer.
+    assert not ha_api._result_is_empty({"rows": [], "count": 40})
+    assert ha_api._result_is_empty({"rows": []})
+    assert not ha_api._result_is_empty({"error": "boom"})

src/vfbquery/ha_api.py

@@ -1354,6 +1354,68 @@ def _slice_page(result, offset=0, page_size=None):
     return page
 
 
+#: Edge-cache lifetime, in seconds, for a result flagged as an underestimate
+#: (see ``vfbquery.solr_result_cache.PARTIAL_RESULT_KEY``). Long enough that a
+#: burst of visitors shares one computation, short enough that the corrected
+#: numbers replace it the same morning rather than next month.
+PARTIAL_RESULT_EDGE_TTL = int(os.getenv("VFBQUERY_PARTIAL_EDGE_TTL", "600") or "600")
+
+#: Same string as ``solr_result_cache.PARTIAL_RESULT_KEY``, repeated here
+#: rather than imported so this module stays free of pysolr at import time.
+PARTIAL_RESULT_KEY = "partial"
+
+
+def result_is_partial(result):
+    """True for a dict result the query function flagged as an underestimate."""
+    return isinstance(result, dict) and bool(result.get(PARTIAL_RESULT_KEY))
+
+
+def _result_is_empty(result):
+    """True for a dict result with no rows to show.
+
+    ``count`` is the authority when present (a paged slice past the end has
+    no rows but a positive count and is not empty); otherwise an empty
+    ``rows`` list decides.
+    """
+    if not isinstance(result, dict):
+        return False
+    count = result.get("count")
+    if isinstance(count, (int, float)) and not isinstance(count, bool):
+        return count == 0
+    rows = result.get("rows")
+    return isinstance(rows, list) and not rows
+
+
+def _edge_cache_headers(result):
+    """Headers telling the v3-cached nginx layer how long to keep *result*.
+
+    That layer ignores ``Cache-Control`` and ``Expires`` (``proxy_ignore_headers``
+    in owl_cache's nginx.conf.template) but honours nginx's own
+    ``X-Accel-Expires``, so this is the one lever the origin has over the
+    edge without an nginx change.
+
+    - An **empty** result (``count`` 0) is never stored at the edge. A true
+      zero costs one origin hit per visitor, answered from the Solr result
+      cache in milliseconds; a false zero — the origin momentarily unable to
+      compute (gamma Kenyon cell, 2026-09-04) — would otherwise be served as
+      "There is no data to display" for ``CACHE_STALE_TIME`` (a month) to
+      everyone.
+    - A **partial** result (flagged by the query function as an
+      underestimate) is kept for :data:`PARTIAL_RESULT_EDGE_TTL` seconds.
+    - Anything else gets no header and the edge's default lifetime.
+
+    ``Cache-Control`` is set alongside for any intermediary that does honour
+    it (browsers, a future edge that stops ignoring it).
+    """
+    if result_is_partial(result):
+        ttl = PARTIAL_RESULT_EDGE_TTL
+        return {"X-Accel-Expires": str(ttl),
+                "Cache-Control": "max-age=%d" % ttl}
+    if _result_is_empty(result):
+        return {"X-Accel-Expires": "0", "Cache-Control": "no-store"}
+    return {}
+
+
 def _page_out(result, func_name, offset=0, page_size=None):
     """Finalise a result for sending: AllAlignedImages is already a single
     server page (just bound it); everything else is sliced from its full set."""
@@ -1462,7 +1524,10 @@ def finish(result):
         out = _page_out(result, func_name, offset, page_size)
         if include_graph:
             out = _maybe_add_graph(out, func_name, short_form)
-        return web.json_response(_with_warnings(out, warnings))
+        # Judge emptiness on the full stored result, not the page: a page past
+        # the end of a non-empty result is not a "no data" answer.
+        return web.json_response(_with_warnings(out, warnings),
+                                 headers=_edge_cache_headers(result))
 
     # Normalize key — AllDatasets ignores the id parameter
     if func_name == "get_all_datasets":

src/vfbquery/solr_result_cache.py

@@ -155,6 +155,16 @@ def cache_doc_glob(namespace: Optional[str] = None) -> str:
 PREVIEW_STATUS_PENDING = 'pending'
 PREVIEW_STATUS_COMPLETE = 'complete'
 
+#: Key a query function sets on a dict result whose numbers are known to be
+#: incomplete (``vfb_queries.PARTIAL_RESULT_KEY`` is the same string). Such a
+#: result is returned to the caller but never written to the cache.
+PARTIAL_RESULT_KEY = 'partial'
+
+
+def result_is_partial(result) -> bool:
+    """True for a dict result flagged as an underestimate."""
+    return isinstance(result, dict) and bool(result.get(PARTIAL_RESULT_KEY))
+
 
 def preview_is_resolved(query: Dict[str, Any]) -> bool:
     """True when a query's preview holds a final answer.
@@ -1423,6 +1433,13 @@ def _call(*call_args, **call_kwargs):
                             full_is_valid = full_result.get('count', -1) >= 0
                         else:
                             full_is_valid = bool(full_result)
+                        if full_is_valid and result_is_partial(full_result):
+                            # An underestimate is served but never stored:
+                            # storing it would make every later caller inherit
+                            # wrong counts with no way to tell.
+                            full_is_valid = False
+                            logger.warning(
+                                f"Not caching partial result for {query_type}({term_id})")
                     elif isinstance(full_result, (list, str)):
                         full_is_valid = len(full_result) > 0
                     else:
@@ -1505,6 +1522,10 @@ def _call(*call_args, **call_kwargs):
                             result_is_error = count_value < 0  # Mark as error if count is negative
                         else:
                             result_is_valid = bool(result)  # For dicts without count field
+                        if result_is_valid and result_is_partial(result):
+                            result_is_valid = False
+                            logger.warning(
+                                f"Not caching partial result for {query_type}({term_id})")
                     elif isinstance(result, (list, str)):
                         result_is_valid = len(result) > 0
                     else: