GSA correctness fixes (review, area 2)

src/gsa/chainlit_flow.py

@@ -13,6 +13,7 @@
 summary at all.
 """
 
+import asyncio
 from pathlib import Path
 from typing import Any, Literal, Protocol, TypedDict
 
@@ -150,7 +151,9 @@ async def _run(
     client: GsaClient | None,
 ) -> None:
     try:
-        matrix = validate(path)
+        # Off the event loop: reading and checking a 20 MB file stalled every
+        # session for seconds (review, area 2).
+        matrix = await asyncio.to_thread(validate, path)
     except UploadRejectedError as refusal:
         await send(str(refusal))
         return
@@ -223,7 +226,12 @@ async def _run(
         return
     except Exception:
         logger.exception("gsa analysis failed", extra={"analysis": analysis_id})
-        await send("Something went wrong while waiting for the analysis.")
+        # With the id: the analysis may well have finished, and the upload is
+        # already deleted, so this is the only way back to it (review, 2).
+        await send(
+            "Something went wrong while waiting for the analysis. It may still "
+            f"finish on Reactome's side, as analysis `{analysis_id}`."
+        )
         return
 
     # `finished.for_model` is deliberately not used here.
@@ -243,6 +251,6 @@ async def _run(
     # way to tell "sent and not shown" from "never sent". This line is the
     # difference.
     logger.info("gsa result sending", extra={"analysis": finished.analysis_id})
-    await send(chat.describe_result(finished))
+    await send(chat.describe_result(finished, grouping))
     await send_file(finished.table_path)
     logger.info("gsa result delivered", extra={"analysis": finished.analysis_id})

src/gsa/chat.py

@@ -114,7 +114,11 @@ def parse_grouping(reply: str, sample_count: int) -> Grouping:
         )
 
     canonical = [seen[label.casefold()] for label in labels]
-    group1, group2 = sorted(seen.values())
+    # The first label to appear is the reference. It was the alphabetical
+    # first, so which group was the baseline -- and so every Up and Down --
+    # depended on spelling: "WT, KO" compared against KO (review, area 2).
+    # People list the control first, as the measured MOCK/MCM run did.
+    group1, group2 = list(seen.values())
     return Grouping(labels=canonical, group1=group1, group2=group2)
 
 
@@ -132,7 +136,7 @@ def describe_progress(status: AnalysisStatus) -> str:
     return f"Running the analysis — {detail}"
 
 
-def describe_result(finished: Finished) -> str:
+def describe_result(finished: Finished, grouping: Grouping | None = None) -> str:
     """What the person reads. Never a prompt.
 
     The Pathway Browser link is included *here* and not in anything the
@@ -148,6 +152,15 @@ def describe_result(finished: Finished) -> str:
         f"**{significant:,} of {total:,} pathways** are significant at FDR < 0.05.",
         "",
     ]
+    if top and grouping is not None:
+        # The table's Direction was never said to be relative to anything.
+        # Only what is certain is stated: which groups, in which order.
+        # ReactomeGSA's own docs do not say which way its sign points.
+        lines += [
+            f"Compared: **{escape(grouping.group1)}** (group 1, the first "
+            f"label you gave) with **{escape(grouping.group2)}** (group 2).",
+            "",
+        ]
     if top:
         lines += [
             "| Pathway | Direction | FDR |",

src/gsa/job.py

@@ -12,18 +12,23 @@
 promises nothing, and `await_result` is where success or failure is decided.
 """
 
+import asyncio
 import contextlib
 import time
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any
+from typing import Any, TypeVar
+
+import httpx
 
 from gsa import results as gsa_results
-from gsa.client import AnalysisStatus, GsaClient, GsaError
+from gsa.client import AnalysisStatus, GsaClient, GsaError, GsaNotReadyError
 from gsa.upload import Matrix, discard
 from util.logging import logging
 
+T = TypeVar("T")
+
 logger = logging.getLogger(__name__)
 
 #: A PADOG run with 1,000 permutations took minutes in the measured run.
@@ -170,7 +175,7 @@ async def submit_uploaded_matrix(
             method=method,
             dataset_name="uploaded",
             dataset_type=dataset_type,
-            matrix=matrix.text,
+            matrix=await asyncio.to_thread(lambda: matrix.text),
             samples=matrix.samples,
             analysis_group=analysis_group,
             group1=group1,
@@ -254,6 +259,35 @@ def prune_results(
     return removed
 
 
+#: Consecutive failures tolerated on one request before giving up, and the
+#: pause between tries. A run takes up to 30 minutes and ~150 polls; with no
+#: tolerance, one 502 or timeout anywhere lost a finished analysis the reader
+#: could not get back (review, area 2: ~17% of 30-minute runs at a 0.1%
+#: per-request failure rate).
+TRANSIENT_ATTEMPTS = 4
+TRANSIENT_PAUSE_SECONDS = 5.0
+
+
+async def _patiently(call: Callable[[], Awaitable[T]]) -> T:
+    """Retry a request through brief outages; a real refusal still raises."""
+    for attempt in range(1, TRANSIENT_ATTEMPTS + 1):
+        try:
+            return await call()
+        except (GsaNotReadyError, AnalysisFailedError):
+            raise
+        except (GsaError, httpx.HTTPError) as exc:
+            if attempt == TRANSIENT_ATTEMPTS:
+                raise
+            logger.warning(
+                "gsa request failed (%s); retrying, attempt %d of %d",
+                type(exc).__name__,
+                attempt,
+                TRANSIENT_ATTEMPTS,
+            )
+            await _sleep(TRANSIENT_PAUSE_SECONDS * attempt)
+    raise AssertionError("unreachable")
+
+
 async def await_result(
     client: GsaClient,
     analysis_id: str,
@@ -275,7 +309,7 @@ async def await_result(
     polls = 0
 
     while True:
-        status = await client.analysis_status(analysis_id)
+        status = await _patiently(lambda: client.analysis_status(analysis_id))
         polls += 1
         if on_progress is not None:
             await on_progress(status)
@@ -294,7 +328,7 @@ async def await_result(
             )
         await _sleep(poll_interval)
 
-    parsed = gsa_results.parse(await client.result(analysis_id))
+    parsed = gsa_results.parse(await _patiently(lambda: client.result(analysis_id)))
     if not parsed.pathways:
         # Complete, and yet nothing to report. Better to say so than to
         # hand back an empty file and an exact-sounding zero.

src/gsa/upload.py

@@ -14,6 +14,8 @@
 """
 
 import contextlib
+import csv
+import io
 import os
 from dataclasses import dataclass
 from pathlib import Path
@@ -46,11 +48,24 @@ class Matrix:
     #: the first thing anyone does with a gene count is show it to someone.
     #: "unknown" is a fact; "-1 genes" is a bug wearing a number.
     gene_count: int | None
+    #: How the upload separates cells; `text` always hands the service tabs.
+    delimiter: str = "\t"
 
     @property
     def text(self) -> str:
-        """The matrix itself. Never logged, never shown, never prompted."""
-        return self.path.read_text()
+        """The matrix, tab-separated. Never logged, never shown, never prompted.
+
+        Decoded the way `validate` decoded it -- a strict read here let a
+        Windows-1252 file pass validation and then fail at submit, every
+        retry (review, area 2) -- and tab-separated whatever was uploaded:
+        the service reads only tabs, and a CSV went through unconverted.
+        A blocking read: call it with `asyncio.to_thread`.
+        """
+        text = decode(self.path.read_bytes())
+        if self.delimiter == "\t":
+            return text
+        rows = csv.reader(io.StringIO(text))
+        return "\n".join("\t".join(cell.strip() for cell in row) for row in rows if row)
 
 
 def max_upload_bytes() -> int:
@@ -64,6 +79,29 @@ def max_upload_bytes() -> int:
     return value if value > 0 else DEFAULT_MAX_UPLOAD_BYTES
 
 
+#: More samples than any expression study here; a bound on what one upload
+#: can make the server hold (a 20 MB one-line header was 7 million names).
+MAX_SAMPLES = 1000
+#: Lines read before deciding it is a matrix, blank ones included: blank
+#: lines used to skip the early exit, and 21 million of them stalled every
+#: session for two seconds.
+MAX_LINES_READ = 5000
+
+
+def decode(raw: bytes) -> str:
+    """The file's text: UTF-8 (with or without BOM), UTF-16, or Windows-1252.
+
+    Excel on Windows writes 1252 or UTF-16; reading those as UTF-8 with
+    replacement showed the reader garbled sample names to label.
+    """
+    if raw.startswith((b"\xff\xfe", b"\xfe\xff")):
+        return raw.decode("utf-16")
+    try:
+        return raw.decode("utf-8-sig")
+    except UnicodeDecodeError:
+        return raw.decode("cp1252", errors="replace")
+
+
 def _split(line: str) -> list[str]:
     # Tab first: it is what the service wants and what every export
     # produces. Comma only if there is no tab at all, because a TSV cell can
@@ -90,22 +128,39 @@ def validate(path: Path) -> Matrix:
         raise UploadRejectedError("That file is empty.")
 
     header: list[str] = []
+    first_row: list[str] = []
     rows = 0
     counted = True
-    with path.open(encoding="utf-8", errors="replace") as handle:
-        for index, line in enumerate(handle):
-            if not line.strip():
-                continue
-            if not header:
-                header = _split(line.rstrip("\n"))
-                continue
-            rows += 1
-            if index > 5000 and rows > MIN_DATA_ROWS:
-                # Enough to know it is a matrix. Counting every gene of a
-                # 20,000-row file to answer "is this a matrix" is work
-                # nobody asked for.
-                counted = False
-                break
+    delimiter = "\t"
+    # Iterated, not split: 21 million blank lines as a list is the problem
+    # this loop's bound exists to avoid.
+    for index, raw_line in enumerate(io.StringIO(decode(path.read_bytes()))):
+        line = raw_line.rstrip("\r\n")
+        if index > MAX_LINES_READ:
+            # Enough to know whether it is a matrix. Counting every gene of
+            # a 20,000-row file is work nobody asked for -- and a file of
+            # blank lines must not make us read all of it.
+            if rows <= MIN_DATA_ROWS:
+                raise UploadRejectedError(
+                    "That file is mostly empty lines. It needs a header row "
+                    "naming the samples, then one row per gene."
+                )
+            counted = False
+            break
+        if not line.strip():
+            continue
+        if not header:
+            delimiter = "\t" if "\t" in line else ","
+            header = _split(line)
+            if len(header) > MAX_SAMPLES + 1:
+                raise UploadRejectedError(
+                    f"That file has {len(header) - 1:,} columns. This handles up "
+                    f"to {MAX_SAMPLES:,} samples."
+                )
+            continue
+        if not first_row:
+            first_row = _split(line)
+        rows += 1
 
     if len(header) < MIN_COLUMNS:
         raise UploadRejectedError(
@@ -118,7 +173,11 @@ def validate(path: Path) -> Matrix:
 
     # The first header cell labels the gene column and is often blank --
     # the measured example's header starts with a tab.
-    samples = [name.strip() for name in header[1:] if name.strip()]
+    # R's write.table writes no cell for the gene column, so the header is
+    # one short of the rows; taking header[1:] then dropped the first
+    # sample, and its label would have gone to the wrong column (review, 2).
+    names = header if len(first_row) == len(header) + 1 else header[1:]
+    samples = [name.strip() for name in names if name.strip()]
     if len(samples) < 2:
         raise UploadRejectedError(
             "There is only one sample in that file. A gene set analysis "
@@ -130,6 +189,7 @@ def validate(path: Path) -> Matrix:
         size_bytes=size,
         samples=samples,
         gene_count=rows if counted else None,
+        delimiter=delimiter,
     )
 
 

tests/gsa/test_gsa_chat.py

@@ -289,3 +289,21 @@ def test_the_matrix_only_reply_offers_no_gene_list() -> None:
     assert "Attach" in chat.HOW_TO_RUN_GSA_WITH_A_MATRIX
     assert "list of genes" not in chat.HOW_TO_RUN_GSA_WITH_A_MATRIX
     assert "list of genes" in chat.HOW_TO_RUN_GSA
+
+
+@pytest.mark.parametrize(
+    ("reply", "reference", "other"),
+    [
+        ("MOCK, MOCK, MCM, MCM", "MOCK", "MCM"),
+        ("WT, KO, WT, KO", "WT", "KO"),
+        ("control, control, Treated, Treated", "control", "Treated"),
+        ("healthy, disease", "healthy", "disease"),
+    ],
+)
+def test_the_first_label_given_is_group_one(
+    reply: str, reference: str, other: str
+) -> None:
+    # It was the alphabetical first, so the baseline depended on spelling
+    # (review, area 2).
+    grouping = chat.parse_grouping(reply, len(reply.split(",")))
+    assert (grouping.group1, grouping.group2) == (reference, other)

tests/gsa/test_gsa_job.py

@@ -17,7 +17,7 @@
 import pytest
 
 from gsa import job
-from gsa.client import AnalysisStatus, DatasetSummary, LoadingStatus
+from gsa.client import AnalysisStatus, DatasetSummary, GsaError, LoadingStatus
 from gsa.upload import Matrix
 
 FIXTURE = Path(__file__).parent / "result_fixture.json"
@@ -472,3 +472,45 @@ async def test_writing_a_result_prunes_the_directory_first(tmp_path: Path) -> No
 
     assert not stale.exists()
     assert finished.table_path.exists()
+
+
+# --- review, area 2 -----------------------------------------------------------
+
+
+class Flaky(StubClient):
+    """Fails the first `failures` status polls, then behaves."""
+
+    def __init__(self, failures: int, **kwargs: Any) -> None:
+        super().__init__(**kwargs)
+        self.failures = failures
+
+    async def analysis_status(self, analysis_id: str) -> AnalysisStatus:
+        if self.failures:
+            self.failures -= 1
+            raise GsaError("GET /status returned 502")
+        return await super().analysis_status(analysis_id)
+
+
+@asyncio_test
+async def test_a_brief_outage_while_waiting_does_not_lose_the_run(
+    tmp_path: Path,
+) -> None:
+    # One 502 anywhere in a 30-minute wait used to end it, and the upload was
+    # already deleted: the finished result was out of reach.
+    finished = await job.await_result(
+        # A fixed count: one derived from TRANSIENT_ATTEMPTS followed a sabotage.
+        Flaky(2),  # type: ignore[arg-type]
+        "an-1",
+        out_dir=tmp_path,
+    )
+    assert finished.analysis_id == "an-1"
+
+
+@asyncio_test
+async def test_a_lasting_outage_still_ends_the_wait(tmp_path: Path) -> None:
+    with pytest.raises(GsaError):
+        await job.await_result(
+            Flaky(job.TRANSIENT_ATTEMPTS),  # type: ignore[arg-type]
+            "an-1",
+            out_dir=tmp_path,
+        )