Failure Detector Expansion 1/5: Read failure type from updated valkey-io/valkey artifact and group nameless failures

scripts/common/issue_dedup.py

@@ -22,8 +22,7 @@
 Search API. The list endpoint draws on the core rate limit (thousands of
 requests per hour) instead of the Search API's 30-per-minute budget, which
 a batch of failures could exhaust, and it is strongly consistent where
-search results can lag the index or silently omit matches. Listings are
-fetched once per publisher and reused across upserts in the same batch.
+search results can lag the index or silently omit matches.
 
 Callers supply rendered title, body, and comment via a render callback;
 this module owns only the dedup machinery. Listing failures are propagated
@@ -52,6 +51,11 @@ class IssueContent:
     body: str
     comment: str
     labels: tuple[str, ...] = ()
+    # Posted as a comment right after the issue is created. The body holds one
+    # trace by design, so a caller with a second one (a valgrind and a sanitizer
+    # report of one bug) puts it here rather than losing it. Empty for the usual
+    # single-trace case, which posts no comment on creation.
+    creation_comment: str = ""
 
 
 class IssueDedupPublisher:
@@ -84,6 +88,13 @@ def __init__(
         self._filter_label = filter_label
         self._open_issues: dict[str, list[Any]] = {}
         self._recently_closed: dict[str, list[Any]] = {}
+        # (repo, issue number) -> body as of this publisher's last update to it.
+        self._bodies: dict[tuple[str, int], str] = {}
+        # Closed issues already used to suppress a creation, by (repo, number).
+        # Keyed the same way as _bodies: upsert takes the repo per call, and
+        # issue numbers restart per repo, so a bare number would let one repo's
+        # suppression block the same number in another.
+        self._suppressed_by_title: set[tuple[str, int]] = set()
 
     def upsert(
         self,
@@ -175,6 +186,20 @@ def upsert(
             # repeated fingerprint or title updates it instead of filing a
             # duplicate. _find_existing above guarantees the cache entry exists.
             self._open_issues[repo_name].append(issue)
+            if content.creation_comment:
+                # Best-effort: the issue exists and records the failure, so a
+                # comment that cannot be posted must not turn a successful
+                # creation into an error.
+                try:
+                    retry_github_call(
+                        lambda: issue.create_comment(body=content.creation_comment),
+                        retries=2, description=f"comment on new issue #{issue.number}",
+                    )
+                except Exception:
+                    logger.warning(
+                        "Could not post the creation comment on issue #%s",
+                        issue.number, exc_info=True,
+                    )
             logger.info("Created issue #%s for %s", issue.number, fingerprint)
             return "created", issue.html_url
 
@@ -199,7 +224,9 @@ def upsert(
         count = int(m.group(1)) + 1 if m else 2
         marker_occurrences = f"<!-- {self._ns}:occurrences:{count} -->"
         new_body = (
-            _occurrence_re(self._ns).sub(marker_occurrences, body)
+            # First occurrence only: the body embeds tool output, and a second
+            # marker pasted into it must not be rewritten as a counter.
+            _occurrence_re(self._ns).sub(marker_occurrences, body, count=1)
             if m else f"{body}\n{marker_occurrences}"
         )
         if idempotency_key is not None:
@@ -213,13 +240,38 @@ def upsert(
             lambda: existing.edit(body=new_body, title=content.title),
             retries=2, description=f"update issue #{existing.number}",
         )
+        # Recorded as soon as the edit lands, before the comment is attempted: a
+        # comment that fails raises out of this call, and the issue is already
+        # claimed by this fingerprint. Recording afterwards left the stale
+        # listing body in place, so a later fingerprint sharing the title
+        # adopted an issue that was no longer unclaimed.
+        self._record_body(repo_name, existing.number, new_body)
         retry_github_call(
             lambda: existing.create_comment(body=content.comment),
             retries=2, description=f"comment on issue #{existing.number}",
         )
         logger.info("Updated issue #%s (occurrence %d)", existing.number, count)
         return "updated", existing.html_url
 
+    def _record_body(self, repo_name: str, number: int, body: str) -> None:
+        """Remember an issue's body as of the update this publisher just made.
+
+        ``edit`` updates only the handle it was called on, so the cached listing
+        entry keeps its pre-update body. A later fingerprint whose title matches
+        would then read the issue as unclaimed and adopt it, overwriting the
+        first failure's marker and filing no issue of its own. Kept beside the
+        listing rather than written onto it: ``Issue.body`` is a read-only
+        property.
+        """
+        self._bodies[(repo_name, number)] = body
+
+    def _body_of(self, repo_name: str, issue: Any) -> str:
+        """An issue's body, preferring one this publisher has since written."""
+        recorded = self._bodies.get((repo_name, issue.number))
+        if recorded is not None:
+            return recorded
+        return issue.body or ""
+
     def _open_issues_for(self, repo: Any, repo_name: str) -> list[Any]:
         """Open issues of the repo, fetched once and cached.
 
@@ -275,19 +327,34 @@ def _recently_closed_for(self, repo: Any, repo_name: str) -> list[Any]:
     def _find_existing(self, repo: Any, repo_name: str, marker: str) -> Any:
         """Find an open issue containing the marker, or None."""
         for issue in self._open_issues_for(repo, repo_name):
-            if marker in (issue.body or ""):
+            if marker in self._body_of(repo_name, issue):
                 return self._reload(repo, issue.number)
         return None
 
     def _find_by_title(self, repo: Any, repo_name: str, title: str) -> Any:
-        """Find an open issue whose title exactly equals ``title``, or None.
+        """Find an open unclaimed issue whose title exactly equals ``title``, or None.
 
-        Migration fallback for when the marker match misses. The comparison
-        is exact and case-sensitive.
+        Migration fallback for when the marker match misses. The comparison is
+        exact and case-sensitive.
+
+        Titles are summarized and truncated, so two different bugs can share
+        one. An issue already stamped with a marker from this namespace belongs
+        to a different fingerprint and must not be adopted: doing so would
+        retarget this fingerprint onto that issue and leave the current failure
+        with no issue of its own.
         """
+        claimed = _fingerprint_marker_re(self._ns)
         for issue in self._open_issues_for(repo, repo_name):
-            if issue.title == title:
-                return self._reload(repo, issue.number)
+            if issue.title != title:
+                continue
+            already_claimed = claimed.search(self._body_of(repo_name, issue))
+            if already_claimed:
+                logger.info(
+                    "Not adopting issue #%s by title: already claimed by fingerprint %s",
+                    issue.number, already_claimed.group(1),
+                )
+                continue
+            return self._reload(repo, issue.number)
         return None
 
     def _find_recently_closed(
@@ -310,15 +377,42 @@ def _find_recently_closed(
         # A legacy issue carries an older (or no) marker, so the marker match
         # misses it. Fall back to an exact title match, mirroring the open-issue
         # migration path, so a just-closed legacy issue is not duplicated.
+        #
+        # The claimed check mirrors _find_by_title and matters more here: titles
+        # are summarized and truncated, so two different bugs can share one, and
+        # suppressing creation is silent. An issue already stamped with a marker
+        # from this namespace belongs to a different fingerprint, so matching it
+        # by title would discard this failure instead of filing it.
         if title_fallback is None:
             return None
+        claimed = _fingerprint_marker_re(self._ns)
         for issue in closed_issues:
-            if issue.title == title_fallback:
+            if issue.title != title_fallback:
+                continue
+            already_claimed = claimed.search(issue.body or "")
+            if already_claimed:
+                logger.info(
+                    "Not suppressing via closed issue #%s: its title matches but "
+                    "it is claimed by fingerprint %s",
+                    issue.number, already_claimed.group(1),
+                )
+                continue
+            # One closed issue stands in for one fingerprint. A title is shared
+            # by more than one bug, and suppression files nothing, so letting it
+            # match repeatedly would drop every fingerprint after the first.
+            if (repo_name, issue.number) in self._suppressed_by_title:
                 logger.info(
-                    "Matched recently closed legacy issue #%s via title fallback",
+                    "Not suppressing via closed issue #%s: already matched by "
+                    "another fingerprint in this run",
                     issue.number,
                 )
-                return issue
+                continue
+            self._suppressed_by_title.add((repo_name, issue.number))
+            logger.info(
+                "Matched recently closed legacy issue #%s via title fallback",
+                issue.number,
+            )
+            return issue
         return None
 
     def _reload(self, repo: Any, number: int) -> Any:
@@ -346,6 +440,29 @@ def _occurrence_re(namespace: str) -> re.Pattern[str]:
     return re.compile(rf"<!-- {re.escape(namespace)}:occurrences:(\d+) -->")
 
 
+def _fingerprint_marker_re(namespace: str) -> re.Pattern[str]:
+    """A fingerprint marker regex: ``<!-- <ns>:<hex> -->``.
+
+    Matches only the hex digest written by ``compute_fingerprint``, so it does
+    not collide with the namespace's other markers (``occurrences``,
+    ``last-key``) or with a legacy marker in some older, non-hex format;
+    those must stay adoptable by title.
+
+    The namespace's own prefix up to its last segment is matched rather than the
+    namespace itself, so an issue claimed under a sibling namespace still reads
+    as claimed. One publisher must not adopt another's issue by title: the
+    namespaces are per-failure-type and titles are summarized and shared across
+    types, so a title collision across two of them would retarget this
+    fingerprint onto an unrelated bug's issue and leave this failure unfiled.
+
+    The segment before the digest must not itself contain a colon, which is what
+    keeps ``last-key`` out: a workflow run id is all digits, so it satisfies the
+    digest pattern and would otherwise make every updated issue read as claimed.
+    """
+    prefix = namespace.rsplit(":", 1)[0] if ":" in namespace else namespace
+    return re.compile(rf"<!-- {re.escape(prefix)}:[^\s:>]*:?([0-9a-f]{{8,}}) -->")
+
+
 def _last_key_marker(namespace: str, key: str) -> str:
     return f"<!-- {namespace}:last-key:{key} -->"
 

scripts/common/workflow_artifacts.py

@@ -15,6 +15,7 @@
 from itertools import islice
 from typing import TYPE_CHECKING, Any
 from urllib.error import HTTPError, URLError
+from urllib.parse import quote
 from urllib.request import Request, urlopen
 
 from scripts.common.github_client import (
@@ -28,6 +29,14 @@
 
 logger = logging.getLogger(__name__)
 
+# Artifacts requested per listing page, GitHub's maximum. Named so the query
+# and the short-page check that ends pagination cannot drift apart.
+_ARTIFACT_PAGE_SIZE = 100
+
+# Bounds the artifact listing so a pathological run can't spin forever. At 100
+# per page this covers far more artifacts than any real run uploads.
+_MAX_ARTIFACT_PAGES = 20
+
 
 @dataclass(frozen=True)
 class WorkflowArtifact:
@@ -60,30 +69,64 @@ def _fetch() -> list[Any]:
             _fetch, retries=self._retries, description=f"list runs {workflow_file}",
         )
 
-    def list_run_artifacts(self, repo_full_name: str, run_id: int) -> list[WorkflowArtifact]:
+    def list_run_artifacts(
+        self, repo_full_name: str, run_id: int, *, name: str | None = None,
+    ) -> list[WorkflowArtifact]:
+        """List a run's artifacts, following pagination.
+
+        A single Valkey Daily run uploads one artifact per job, hundreds after
+        matrix expansion, so the response is paginated and reading only the
+        first page can miss the artifact entirely. Pass ``name`` to have GitHub
+        filter server-side, which is both cheaper and immune to that ordering.
+        """
         repo = self._gh.get_repo(repo_full_name)
+        base = f"/repos/{repo_full_name}/actions/runs/{run_id}/artifacts"
+        query = f"per_page={_ARTIFACT_PAGE_SIZE}"
+        if name is not None:
+            query += f"&name={quote(name)}"
 
-        def _fetch() -> Any:
-            _, data = repo._requester.requestJsonAndCheck(
-                "GET", f"/repos/{repo_full_name}/actions/runs/{run_id}/artifacts",
+        artifacts: list[WorkflowArtifact] = []
+        seen_ids: set[int] = set()
+        for page in range(1, _MAX_ARTIFACT_PAGES + 1):
+            def _fetch(page: int = page) -> Any:
+                _, data = repo._requester.requestJsonAndCheck(
+                    "GET", f"{base}?{query}&page={page}",
+                )
+                return data
+
+            payload = retry_github_call(
+                _fetch, retries=self._retries,
+                description=f"list artifacts {run_id} page {page}",
             )
-            return data
-
-        payload = retry_github_call(_fetch, retries=self._retries,
-                                    description=f"list artifacts {run_id}")
-        if not isinstance(payload, dict):
-            return []
-        return [
-            WorkflowArtifact(
-                artifact_id=a["id"], name=a["name"],
-                size_in_bytes=a.get("size_in_bytes", 0),
-                expired=a.get("expired", False),
+            if not isinstance(payload, dict):
+                break
+            entries = payload.get("artifacts")
+            if not isinstance(entries, list) or not entries:
+                break
+            for a in entries:
+                if (
+                    isinstance(a, dict)
+                    and isinstance(a.get("id"), int)
+                    and isinstance(a.get("name"), str)
+                    and a["id"] not in seen_ids
+                ):
+                    seen_ids.add(a["id"])
+                    artifacts.append(WorkflowArtifact(
+                        artifact_id=a["id"], name=a["name"],
+                        size_in_bytes=a.get("size_in_bytes", 0),
+                        expired=a.get("expired", False),
+                    ))
+            total = payload.get("total_count")
+            if isinstance(total, int) and len(seen_ids) >= total:
+                break
+            if len(entries) < _ARTIFACT_PAGE_SIZE:
+                break
+        else:
+            logger.warning(
+                "Stopped listing artifacts for run %d at the %d-page cap; "
+                "%d seen so far", run_id, _MAX_ARTIFACT_PAGES, len(artifacts),
             )
-            for a in payload.get("artifacts", [])
-            if isinstance(a, dict)
-            and isinstance(a.get("id"), int)
-            and isinstance(a.get("name"), str)
-        ]
+        return artifacts
 
     def download_artifact(
         self,