diff --git a/scripts/common/issue_dedup.py b/scripts/common/issue_dedup.py index 93e8506c..a37ed6c9 100644 --- a/scripts/common/issue_dedup.py +++ b/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"" 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,6 +240,12 @@ 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}", @@ -220,6 +253,25 @@ def upsert( 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"") +def _fingerprint_marker_re(namespace: str) -> re.Pattern[str]: + """A fingerprint marker regex: ````. + + 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"") + + def _last_key_marker(namespace: str, key: str) -> str: return f"" diff --git a/scripts/common/workflow_artifacts.py b/scripts/common/workflow_artifacts.py index ba1fb93e..882acddf 100644 --- a/scripts/common/workflow_artifacts.py +++ b/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, diff --git a/scripts/test_failure_detector/download.py b/scripts/test_failure_detector/download.py index b116d681..bff29bc3 100644 --- a/scripts/test_failure_detector/download.py +++ b/scripts/test_failure_detector/download.py @@ -4,6 +4,9 @@ import logging import re +from dataclasses import dataclass, field +from itertools import islice +from typing import Any from github import Github from github.WorkflowRun import WorkflowRun @@ -17,6 +20,19 @@ _FAILURES_JSON_NAME = "all-test-failures.json" _FAILURES_ARTIFACT_NAME = "all-test-failures" +# How many runs to scan for the newest usable one. Only pull_request runs sit +# between nightly crons, a handful a day at most, so this reaches well past +# yesterday's schedule run; beyond it, reporting nothing found beats paging +# through the whole history. +_MAX_RUNS_SCANNED = 50 + +# Only the nightly schedule is discovered automatically. A dispatched Daily is +# not the same run: its inputs can skip jobs and point the checkout at another +# repository or ref, so its failures need not belong to this branch's code, and +# a partial run's absent jobs would read as passing. A maintainer who does want +# one analyzed can name it with --run-id, which bypasses this filter. +_ANALYZABLE_EVENTS = frozenset({"schedule"}) + def get_latest_daily_run( gh: Github, repo_full_name: str, @@ -46,23 +62,43 @@ def get_latest_daily_run( logger.warning("Workflow %r not found in %s", workflow_name, repo_full_name) return None - # Restrict to scheduled runs. The Valkey Daily workflow also runs on - # pull_request (and fork PRs sit at action_required with no artifacts), - # so we'd sometimes silently analyze the wrong run. + # Events are filtered locally, alongside the conclusion check below, so both + # rejection reasons are visible in one place and log the run they skipped. + # The scan is bounded instead: pull_request runs interleave with the nightly + # cron, so the newest schedule run sits a few entries down. + # + # get_runs() is lazy, so the islice must run inside the retried call for the + # retries to cover the actual request. runs = retry_github_call( - lambda: daily_workflow.get_runs( - branch=branch, status="completed", event="schedule", - ), + lambda: list(islice( + daily_workflow.get_runs(branch=branch, status="completed"), + _MAX_RUNS_SCANNED, + )), retries=3, description=f"list runs for {workflow_name}", ) for run in runs: + # The Daily workflow also runs on pull_request. Such a run tests the + # PR's merge commit, not the branch, so its failures belong to the PR + # and filing them as branch failures would blame the wrong code. Most + # sit at action_required and would be dropped below anyway, but a PR + # from a branch in the same repo runs without approval and reaches a + # real conclusion, so the event must be checked explicitly. + if run.event not in _ANALYZABLE_EVENTS: + logger.debug( + "Skipping run #%d (event=%s)", run.run_number, run.event, + ) + continue # Skip runs that never actually executed: cancelled/skipped, runs - # awaiting approval (action_required, e.g. fork PRs) or expired - # (stale), and runs with no conclusion yet. These produce no test + # awaiting approval (action_required, e.g. fork PRs), expired (stale), + # runs that died before any job started (startup_failure, e.g. invalid + # workflow YAML), and runs with no conclusion yet. These produce no test # artifacts and would be mistaken for a clean pass. - if run.conclusion in ("cancelled", "skipped", "action_required", "stale", None): + if run.conclusion in ( + "cancelled", "skipped", "action_required", "stale", + "startup_failure", None, + ): logger.debug( "Skipping run #%d (conclusion=%s)", run.run_number, run.conclusion, ) @@ -99,21 +135,27 @@ def download_all_test_failures( """ client = artifact_client or ArtifactClient(gh, token=github_token) - artifacts = client.list_run_artifacts(repo_full_name, run_id) - target = next( - (a for a in artifacts if a.name == _FAILURES_ARTIFACT_NAME), None + artifacts = client.list_run_artifacts( + repo_full_name, run_id, name=_FAILURES_ARTIFACT_NAME, ) - if target is None: + matches = [a for a in artifacts if a.name == _FAILURES_ARTIFACT_NAME] + if not matches: logger.info( "No %r artifact found in run %d", _FAILURES_ARTIFACT_NAME, run_id ) return None - if target.expired: + + # Re-running a workflow leaves one artifact per attempt under the same run + # and name, each expiring on its own clock. Take the newest live one: a + # stale earlier attempt must not shadow the re-run's usable artifact. + live = [a for a in matches if not a.expired] + if not live: logger.warning( - "Artifact %r (id=%d) in run %d has expired", - target.name, target.artifact_id, run_id, + "All %d %r artifact(s) in run %d have expired", + len(matches), _FAILURES_ARTIFACT_NAME, run_id, ) return None + target = max(live, key=lambda a: a.artifact_id) logger.info("Downloading artifact: %s (id=%d)", target.name, target.artifact_id) files = client.download_artifact( @@ -131,15 +173,120 @@ def download_all_test_failures( logger.info("Extracted %s from artifact zip", _FAILURES_JSON_NAME) return content -def get_job_urls( +def get_run_conclusion( gh: Github, repo_full_name: str, run_id: int, -) -> dict[str, str]: - """Get a mapping of job name -> HTML URL for all jobs in a workflow run. +) -> str | None: + """A workflow run's conclusion, or None if it cannot be determined. - Also includes normalized variants (parentheses replaced with dashes, - spaces replaced with dashes) for fuzzy matching. + Used when the run was named explicitly rather than discovered, so the + caller can still tell a red run apart from a clean one. Returns None on + any API failure: the conclusion only sharpens an error message, so it must + not turn a usable run into a hard failure. + """ + try: + repo = retry_github_call( + lambda: gh.get_repo(repo_full_name), + retries=3, + description=f"get repo {repo_full_name}", + ) + run = retry_github_call( + lambda: repo.get_workflow_run(run_id), + retries=3, + description=f"get run {run_id}", + ) + except Exception: + logger.warning( + "Could not fetch conclusion for run %d", run_id, exc_info=True, + ) + return None + return run.conclusion + + +@dataclass(frozen=True) +class JobInfo: + """Job metadata derived from a workflow run's job list. + + ``urls`` maps job name (and normalized aliases) to the job's HTML URL. + ``step_urls`` maps job name -> suite -> a URL anchored to the step that + ran that suite, so a failure links to its own step rather than the job's + first failed step. ``failed`` holds the names of jobs that failed. + """ + + urls: dict[str, str] + failed: set[str] + step_urls: dict[str, dict[str, str]] = field(default_factory=dict) + + def url_for(self, job_name: str, suite: str) -> str: + """URL for a suite's failure in a job, anchored to its step if known. + + Falls back to the plain job URL when the suite's step cannot be + identified (an unmapped suite, or a job whose steps were unavailable). + """ + step_url = self.step_urls.get(job_name, {}).get(suite) + return step_url or self.urls.get(job_name, "") + + +# Job conclusions that mean the job did not pass and so may hold a failure the +# artifact never captured. A job the runner killed on the job timeout concludes +# "timed_out" rather than "failure", and that is exactly the case whose console +# log still carries the [TIMEOUT] lines timeout recovery reads, so leaving it out +# skipped the jobs the scan exists for. +_FAILED_JOB_CONCLUSIONS = frozenset({"failure", "timed_out"}) + +# The Daily workflow runs each test suite in a named step. A suite's failures +# live in .json, so link a failure to the step that produced it instead +# of the job's first failed step. Keys are artifact suite names; values match a +# step name. gtest failures come from the unittest step but are extracted in a +# following step, so both spellings map to the unittest suite. +_SUITE_STEP_NAMES: dict[str, str] = { + "valkey": "test", + "moduleapi": "module api test", + "sentinel": "sentinel tests", + "unittest": "unittest", +} + + +def _suite_step_number(suite: str, steps: list[Any]) -> int | None: + """Return the 1-based step number that ran ``suite``, or None. + + Matches the mapped step name case-insensitively. Returns None for a suite + with no mapping or when no step name matches, so the caller keeps the plain + job URL. + """ + step_name = _SUITE_STEP_NAMES.get(suite) + if step_name is None: + return None + for step in steps: + if step.name and step.name.lower() == step_name.lower(): + return step.number + return None + + +def normalize_job_name(job_name: str) -> str: + """Convert an API job name to the spelling the artifact uses. + + A matrix job is named ``base (value)`` by the API, but its artifact is + uploaded as ``base-value`` because the workflow interpolates the matrix + variable into ``job-name``. Callers matching one against the other must + normalize first. + """ + collapsed = re.sub(r"\s*\(([^)]+)\)", r"-\1", job_name) + return re.sub(r"\s+", "-", collapsed) + + +def get_job_info( + gh: Github, + repo_full_name: str, + run_id: int, +) -> JobInfo: + """Fetch job metadata for a workflow run in a single API call. + + Returns a :class:`JobInfo` containing: + - ``urls``: job name -> HTML URL (includes normalized aliases for fuzzy + matching against artifact names). + - ``failed``: names of jobs whose conclusion indicates failure. """ repo = retry_github_call( @@ -154,28 +301,68 @@ def get_job_urls( description=f"get run {run_id}", ) - jobs = retry_github_call( - lambda: run.jobs(), + # The list() must happen inside the retried call: jobs() returns a lazy + # PaginatedList that issues no request until iterated, so retrying only the + # construction would leave the actual HTTP call unprotected. + job_list = retry_github_call( + lambda: list(run.jobs()), retries=3, description=f"list jobs for run {run_id}", ) - # Materialize once: jobs may be a lazy paginated list, and we iterate twice. - job_list = list(jobs) - - # First pass: exact job names. These are authoritative, so they take - # precedence over any normalized alias. job_url_map: dict[str, str] = {job.name: job.html_url for job in job_list} + failed_jobs: set[str] = set() + step_url_map: dict[str, dict[str, str]] = {} - # Second pass: normalized variants for fuzzy matching against artifact - # names. Only add an alias when it does not collide with an exact job name, - # so a normalized alias of one job can never overwrite another job's exact - # mapping and attach the wrong CI URL. for job in job_list: - normalized = re.sub(r"\s*\(([^)]+)\)", r"-\1", job.name) - normalized = re.sub(r"\s+", "-", normalized) + normalized = normalize_job_name(job.name) + if job.conclusion in _FAILED_JOB_CONCLUSIONS: + # Both spellings are recorded. The API names a matrix job + # "base (value)" while the artifact keys it "base-value", and + # callers join on one or the other: log recovery looks up the + # artifact's key, enrichment intersects the failure's job names + # (artifact spelling) with this set. Holding only the API name left + # every matrix job out of that intersection, so a matrix job's + # gtest and timeout placeholders were never enriched. + failed_jobs.add(job.name) + failed_jobs.add(normalized) + if normalized != job.name and normalized not in job_url_map: job_url_map[normalized] = job.html_url - logger.info("Found %d job URL mappings for run %d", len(job_url_map), run_id) - return job_url_map + # Anchor each suite to the step that ran it. The list endpoint returns + # steps inline, so this costs no extra API call. A job with no matching + # steps contributes nothing and keeps the plain job URL. + suite_urls: dict[str, str] = {} + for suite in _SUITE_STEP_NAMES: + number = _suite_step_number(suite, job.steps) + if number is not None: + suite_urls[suite] = f"{job.html_url}#step:{number}:1" + if suite_urls: + step_url_map[job.name] = suite_urls + if normalized != job.name: + step_url_map.setdefault(normalized, suite_urls) + + logger.info( + "Found %d job URL mappings (%d failed) for run %d", + len(job_url_map), len(failed_jobs), run_id, + ) + return JobInfo(urls=job_url_map, failed=failed_jobs, step_urls=step_url_map) + + +def get_job_urls( + gh: Github, + repo_full_name: str, + run_id: int, +) -> dict[str, str]: + """Get a mapping of job name -> HTML URL for all jobs in a workflow run. + + Also includes normalized variants (parentheses replaced with dashes, + spaces replaced with dashes) for fuzzy matching. + + Timeout recovery also needs to know which jobs failed, so the detector + calls :func:`get_job_info` and reads both fields off one response rather + than paying for a second job listing. This narrower view is kept for + callers that only want the URLs. + """ + return get_job_info(gh, repo_full_name, run_id).urls diff --git a/scripts/test_failure_detector/issue_renderer.py b/scripts/test_failure_detector/issue_renderer.py index 2fde30f4..4621d98f 100644 --- a/scripts/test_failure_detector/issue_renderer.py +++ b/scripts/test_failure_detector/issue_renderer.py @@ -1,15 +1,17 @@ """Render detected test failures into GitHub issue title, body, and comment text. -The rendering is test-failure-specific (test name/file, error trace, the list -of CI jobs the failure appeared in); the create-or-update machinery lives in +Supports multiple failure types (assertion, sanitizer, valgrind, timeout, +exception, startup, memory-leak, unittest) with type-specific titles, labels, +and fingerprint namespaces. The create-or-update machinery lives in :mod:`scripts.common.issue_dedup`. -A test failure's identity is the ``test_name`` + ``test_file`` pair, which is -the dedup fingerprint. Across recurrences we accumulate the set of failing -environments (CI jobs) into the issue body; because the dedup publisher's -``render`` callback can't see the previously published body, that merge is done -via the publisher's ``body_transform`` hook (see -:meth:`_FailureRenderer.merge_environments`). +For failures WITH a test_name (assertions, timeouts, gtest), the identity is +the (type, test_name, test_file) triple. For failures WITHOUT a test_name +(sanitizer/valgrind/startup), the identity is (type, normalized_error), so the +same underlying bug produces one issue regardless of which test file triggered +the detection. Titles follow the identity: the test file appears only for types +whose fingerprint keys on it, so a title is not rewritten when the same bug is +detected under a different file. """ from __future__ import annotations @@ -19,70 +21,105 @@ from scripts.common.incidents import compute_fingerprint from scripts.common.issue_dedup import IssueContent -from scripts.test_failure_detector.parse_failures import UniqueFailure +from scripts.test_failure_detector.parse_failures import ( + TIMESTAMP_PATTERNS, + FailureType, + UniqueFailure, + is_plumbing_frame, + normalize_error_identity, + scrub_volatile_tokens, +) MARKER_NAMESPACE = "valkey-ci-agent:test-failure" LABEL_NAME = "test-failure" +# Type-specific marker namespaces for fingerprinting and issue search. +_TYPE_NAMESPACE: dict[FailureType, str] = { + FailureType.ASSERTION: "valkey-ci-agent:test-failure", + FailureType.SANITIZER: "valkey-ci-agent:sanitizer-error", + FailureType.VALGRIND: "valkey-ci-agent:valgrind-error", + FailureType.TIMEOUT: "valkey-ci-agent:test-timeout", + FailureType.STARTUP: "valkey-ci-agent:startup-failure", + FailureType.EXCEPTION: "valkey-ci-agent:test-exception", + FailureType.MEMORY_LEAK: "valkey-ci-agent:memory-leak", + FailureType.UNITTEST: "valkey-ci-agent:unittest-failure", +} + +# Every type shares one title prefix. The failure type is named in the body +# instead, so an issue is not retitled when the same bug is later attributed to +# a different type (a valgrind report and a sanitizer report of one bug). +TITLE_PREFIX = "[TEST-FAILURE]" + + +def marker_namespace_for(failure: UniqueFailure) -> str: + """Return the marker namespace for a failure's type.""" + return _TYPE_NAMESPACE.get(failure.failure_type, MARKER_NAMESPACE) -def fingerprint_for(failure: UniqueFailure) -> str: - """Stable dedup key for a failure: a hash of test name + file. - The identity is ``test_name`` + ``test_file``, hashed via - :func:`scripts.common.incidents.compute_fingerprint` like the fuzzer - pipeline, into a fixed-shape hex token safe for the HTML-comment marker. +def label_for(failure: UniqueFailure) -> str: + """Return the issue label. All failure types use the same label. - The pair is the identity, so it goes in ``namespace`` (joined in order, - never normalized) rather than ``shapes``. That keeps digits significant so - PSYNC2 and PSYNC3 stay distinct, and preserves order so a name/file swap - cannot collide. + One label keeps the tracker filter simple and needs no new labels created on + the target repo. The type is named in the body instead. """ - return compute_fingerprint( - namespace=(MARKER_NAMESPACE, failure.test_name, failure.test_file), - shapes=(), - ) + return LABEL_NAME -def title_for(failure: UniqueFailure) -> str: - """Issue title for a failure. +def fingerprint_for(failure: UniqueFailure) -> str: + """Stable dedup key for a failure. + + Keyed on (test_name, test_file) when the failure names a test, on test_file + for a nameless timeout (every timeout shares one generic error text), and on + the normalized error otherwise. + + Identity components go in ``namespace``, never ``shapes``: shapes collapses + every run of digits, which would merge PSYNC2 with PSYNC3 and a + use-after-free at cluster_legacy.c:3421 with one at :5109. - Exposed rather than inlined in the renderer so callers can pass the same - title to ``IssueDedupPublisher.upsert`` as ``title_fallback`` when - migrating issues off the old raw-fingerprint marker. """ + ns = marker_namespace_for(failure) + + if failure.has_test_identity: + return compute_fingerprint( + namespace=(ns, failure.test_name, failure.test_file), + shapes=(), + ) + elif failure.failure_type == FailureType.TIMEOUT and failure.test_file: + return compute_fingerprint( + namespace=(ns, failure.test_file), + shapes=(), + ) + else: + error_identity = normalize_error_identity(failure.error) + return compute_fingerprint( + namespace=(ns, error_identity), + shapes=(), + ) + + +def title_for(failure: UniqueFailure) -> str: + """Issue title for a failure.""" return _build_title(failure) def renderer_for(failure: UniqueFailure) -> _FailureRenderer: """Return a renderer supplying the ``render`` and ``body_transform`` hooks that :class:`IssueDedupPublisher.upsert` expects for one failure. - - The two hooks are coupled so the recurrence comment can name the *newly* - failing environments. ``upsert`` runs ``body_transform`` (which diffs the - failure's environments against the previously published body) before - ``render`` (which builds the comment), so by the time the comment is - rendered the renderer already knows which environments were not recorded - before. See :meth:`_FailureRenderer.merge_environments`. """ return _FailureRenderer(failure) class _FailureRenderer: - """Per-failure ``render``/``body_transform`` pair sharing the set of newly - failing environments. Created via :func:`renderer_for`.""" + """Per-failure render/body_transform pair. Created via :func:`renderer_for`.""" def __init__(self, failure: UniqueFailure) -> None: self._failure = failure - # Environments failing for the first time on this run, populated by - # ``merge_environments`` on the update path. Empty on the create path - # (no prior body to diff, and no comment is posted there anyway). self._newly_failing: list[str] = [] - # The latest error trace when it differs (ignoring run-specific noise) - # from the one recorded on the issue, populated by ``merge_environments`` - # on the update path. ``None`` when the trace is unchanged or absent, so - # :meth:`render` only calls it out when there is something new to show. self._new_error: str | None = None + # Digests of the traces _detect_new_error selected for publication + # this run. Only these are added to the issue's record. + self._published_digests: list[str] = [] def render(self, marker: str, occurrences: int) -> IssueContent: """The ``render`` callback: title/body/comment/labels for the issue.""" @@ -94,95 +131,692 @@ def render(self, marker: str, occurrences: int) -> IssueContent: newly_failing=self._newly_failing, new_error=self._new_error, ), - labels=(LABEL_NAME,), + labels=(label_for(self._failure),), + creation_comment=_build_absorbed_trace_comment(self._failure), ) def merge_environments(self, existing_body: str) -> str: """The ``body_transform`` callback: fold this failure's environments into the existing issue body, preserving environments recorded by - earlier runs and recording which ones are newly failing so - :meth:`render` can call them out in the recurrence comment. - - Also diffs the failure's error trace against the one recorded on the - issue and, when it has meaningfully changed, records it so - :meth:`render` can surface the new trace in the comment. The body's - original trace is left intact — the body is the first-seen record, the - comment timeline carries each subsequent change. + earlier runs and recording which ones are newly failing. + + Also records which tools' traces the issue has reported, so a trace + published in a comment is not published again on every later run. The + recorded traces are a marker, not content: the body's own trace section + is left as first published. + + Only the traces this run actually published are recorded, keyed by + content. Recording every trace the failure carries would suppress one + that later changes. """ self._new_error = self._detect_new_error(existing_body) - existing_envs = _extract_environments_from_body(existing_body) + body = existing_body + if self._published_digests: + body = _record_reported_tools(body, self._published_digests) + existing_envs = _extract_environments_from_body(body) self._newly_failing = [ j.job for j in self._failure.jobs if j.job not in existing_envs ] if not self._newly_failing: - return existing_body + return body return _update_environments_in_body( - existing_body, existing_envs + self._newly_failing, + body, existing_envs + self._newly_failing, ) def _detect_new_error(self, existing_body: str) -> str | None: - """Return the failure's error trace when it differs from the one stored - on the issue, else ``None``. - - The comparison is normalized (see :func:`_normalize_trace`) so that - run-specific noise — timestamps, ports/PIDs, hex addresses, temp paths — - does not flag an unchanged failure as new on every recurrence. An empty - new error is never called out. - - Legacy issues predating the Error stack trace section have no stored - trace (``_extract_error_from_body`` returns ``""``); with no baseline to - diff against, the trace is not called out. Otherwise the body never - backfilled with the section would diff against "" and re-post the same - "new" trace on every recurrence. + """Return this run's traces that are not already recorded on the issue, + or None when it carries nothing new. + + Every trace the failure holds is considered, not just its own: a failure + that absorbed another tool's report carries both, and the absorbed one is + the whole point of the merge. The update path publishes only the body it + was given plus this comment, so a trace the issue lacks reaches a reader + here or not at all. + + A trace matching any stored one says nothing new. One matching none is + new, whether because it changed or because that tool's report has not + appeared on this issue before. """ - new_error = self._failure.error - if not new_error.strip(): - return None - stored = _extract_error_from_body(existing_body) - if not stored.strip(): + stored = [ + _normalize_trace(t) + for t in _extract_errors_from_body(existing_body) + if t.strip() + ] + if not stored: return None - if _normalize_trace(stored) == _normalize_trace(new_error): + + # A tool whose trace was already published in an earlier comment is not + # published again: the body keeps only its first-seen trace, so without + # this the same comment would repeat on every later run. + reported = _reported_tools_in_body(existing_body) + traces = [ + (trace_label_for(self._failure), self._failure.error), + *self._failure.extra_traces, + ] + fresh: list[str] = [] + published: list[str] = [] + for _label, trace in traces: + # The stored traces were truncated when published, so a fresh trace + # must be compared in its truncated form too, or every recurrence of + # an oversized trace would register as new. The budget must be the + # one the body used, not a share of it: comparing a half-budget + # candidate against a full-budget stored trace never matches. The + # backtick bounding the renderers apply has to be matched here too, + # for the same reason: the body holds the bounded form. + candidate = _bound_backtick_runs( + _truncate_trace(trace, _MAX_TRACE_CHARS) + ) + if not candidate.strip(): + continue + digest = trace_digest(candidate) + if digest in reported: + continue + if _normalize_trace(candidate) in stored: + continue + fresh.append(candidate) + if digest not in published: + published.append(digest) + self._published_digests = published + if not fresh: return None - return new_error + # Joined as one trace: the comment renders it under the template's + # heading, which carries no per-tool label. + return "\n\n".join(fresh) + + +# Keywords that mark a line as carrying diagnostic content rather than +# boilerplate. Used by _error_summary_line to prefer the real payload over +# the generic runner prefix/banner. +_TITLE_KEYWORDS = ( + "Invalid", "definitely lost", "indirectly lost", + "heap-buffer-overflow", "heap-use-after-free", + "stack-buffer-overflow", "use-after-poison", + # Valgrind spells it the British way; the sanitizers do not. + "uninitialized", "uninitialised", "runtime error", + "detected memory leaks", "LEAK SUMMARY", + "fishy", "overlap", "Mismatched", + "possibly lost", "Jump to the invalid address", + "Process terminating with default action of signal", +) + + +# Heap-layout coordinates in a diagnostic line ("in loss record 900 of +# 1,109") shift between runs of the same bug. The fingerprint already +# scrubs them; the title must too, or each recurrence rewrites the title +# of the same issue. +_LOSS_RECORD_RE = re.compile(r"\s*\bin loss record \d[\d,]* of \d[\d,]*") + +# Allocation sizes drift run to run for the same leak ("49 bytes" vs +# "52 bytes"), so titles show them as N: "N bytes in N blocks are +# definitely lost". +_COUNT_RE = re.compile(r"\b\d[\d,]*(\s+(?:bytes?|blocks?|byte\(s\)|object\(s\)))\b") + +# An AddressSanitizer diagnostic line ends in a volatile address dump +# ("heap-use-after-free on address 0x60... at pc 0x... bp 0x... sp 0x..."). The +# address and registers change every run of the same bug; the fingerprint +# scrubs them, so the title must too or it is rewritten on each recurrence. +_ASAN_ADDR_NOISE_RE = re.compile(r"\s+on address 0x[0-9a-fA-F]+.*$") + +# Volatile run-specific tokens in generic title candidates: ports, PIDs, hex +# addresses, temp paths, timestamps, and long bare numbers. The publisher +# re-titles on every update, so a token the fingerprint scrubs but the title +# keeps rewrites one issue's title on each recurrence. Anything added to +# scrub_volatile_tokens for identity needs a counterpart here. +_TITLE_VOLATILE_SUBS: tuple[tuple[re.Pattern, str], ...] = ( + (re.compile(r"0x[0-9a-fA-F]+"), "0xN"), + (re.compile(r"/tmp/[^\s:]+"), "/tmp/..."), + (re.compile(r"\b(pid|port)([=\s]+)\d+", re.IGNORECASE), r"\1\2N"), + # A hung client's last-known runner state ("last state: (SPAWNING SERVER) + # pid:N fd 12"). It names what the runner was doing, not the failure, and + # every token in it drifts. The body keeps it; the title must not. + (re.compile(r"[,;]?\s*last state:.*$", re.IGNORECASE | re.DOTALL), ""), + # The timestamps the identity drops, shared from there so the two cannot + # drift apart: a datestamp the title keeps but the identity scrubs retitles + # one issue on every recurrence. + *((pattern, "