Run build repair only on regressions

services/hackbot-pulse-listener/app/__main__.py

@@ -19,7 +19,7 @@ def main() -> None:
         logger.warning("PULSE_USER/PULSE_PASSWORD not set; listener will not start")
         return
 
-    executor = ThreadPoolExecutor(max_workers=settings.poll_max_workers)
+    executor = ThreadPoolExecutor(max_workers=settings.max_workers)
     consumer_obj = consumer.build_consumer(executor)
 
     def shutdown(signum, _frame):

services/hackbot-pulse-listener/app/config.py

@@ -32,7 +32,12 @@ class Settings(BaseSettings):
     # Polling the API for run completion
     poll_interval_seconds: int = 60
     run_max_age_minutes: int = 12 * 60
-    poll_max_workers: int = 8
+    # Shared worker pool for message processing and run polling. A
+    # regression check may block for a few minutes waiting for a parent
+    # build to settle, so the pool is sized well above the number of
+    # builds/runs in flight at once. Threads are cheap and mostly idle
+    # while waiting.
+    max_workers: int = 256
 
     # Email notifications (SendGrid)
     sendgrid_api_key: str | None = None

services/hackbot-pulse-listener/app/consumer.py

@@ -1,11 +1,12 @@
 import logging
+import threading
 from concurrent.futures import Executor
 
 from cachetools import TTLCache
 from kombu import Connection, Exchange, Queue
 from kombu.mixins import ConsumerMixin
 
-from app import client, lando, taskcluster, worker
+from app import client, lando, regression, taskcluster, worker
 from app.config import settings
 from app.models import RunContext
 
@@ -15,11 +16,16 @@
 
 EXCHANGES = ("exchange/taskcluster-queue/v1/task-failed",)
 
-# In-memory dedupe of hg revisions already handed to the agent. Only the
-# single consumer thread touches it, so no lock is needed.
+# In-memory dedupe of hg revisions already handed to the agent. A revision is
+# recorded only once we actually trigger a run, so an inherited failure on one
+# build label never suppresses a genuine regression on another label of the
+# same push, while a revision that breaks many builds still triggers only once.
+# Messages are handled on worker threads, so the check-and-record is done under
+# a lock.
 _seen: TTLCache = TTLCache(
     maxsize=settings.dedupe_max_size, ttl=settings.dedupe_ttl_seconds
 )
+_seen_lock = threading.Lock()
 
 
 def process(body: dict, executor: Executor) -> str | None:
@@ -43,10 +49,25 @@ def process(body: dict, executor: Executor) -> str | None:
         logger.warning("No GECKO_HEAD_REV for task %s; skipping", task_id)
         return None
 
-    if hg_revision in _seen:
-        logger.info("Revision %s already processed; skipping", hg_revision)
+    with _seen_lock:
+        already_seen = hg_revision in _seen
+    if already_seen:
+        logger.info("Revision %s already triggered a run; skipping", hg_revision)
         return None
-    _seen[hg_revision] = True
+
+    if not regression.is_new_build_failure(project, hg_revision, task_label):
+        logger.info(
+            "Build %s at %s inherited from an ancestor push; skipping",
+            task_label,
+            hg_revision,
+        )
+        return None
+
+    with _seen_lock:
+        if hg_revision in _seen:
+            logger.info("Revision %s already triggered a run; skipping", hg_revision)
+            return None
+        _seen[hg_revision] = True
 
     git_commit = lando.hg_to_git(hg_revision)
     if not git_commit:
@@ -56,7 +77,8 @@ def process(body: dict, executor: Executor) -> str | None:
             task_id,
             project,
         )
-        _seen.pop(hg_revision, None)
+        with _seen_lock:
+            _seen.pop(hg_revision, None)
         return None
 
     inputs: dict = {
@@ -73,7 +95,8 @@ def process(body: dict, executor: Executor) -> str | None:
         run_id = client.trigger_run(inputs)
     except Exception:
         logger.exception("Failed to trigger build-repair run for %s", hg_revision)
-        _seen.pop(hg_revision, None)
+        with _seen_lock:
+            _seen.pop(hg_revision, None)
         return None
 
     logger.info(
@@ -97,24 +120,38 @@ def process(body: dict, executor: Executor) -> str | None:
 
 
 def make_handler(executor: Executor):
-    def on_message(body, message):
+    def run(body: dict) -> None:
         try:
             process(body, executor)
         except Exception:
             logger.exception("Error handling pulse message")
+
+    def on_message(body, message):
+        # Process on a worker thread so a regression check that blocks waiting
+        # for a parent build to settle never stalls the consumer thread.
+        try:
+            executor.submit(run, body)
+        except Exception:
+            logger.exception("Failed to dispatch pulse message")
         finally:
             message.ack()
 
     return on_message
 
 
 def _build_queues(user: str) -> list[Queue]:
+    # Both local and prod authenticate as the same pulse user, so the queue name
+    # must also vary by environment; otherwise both consumers bind to the same
+    # durable queue and steal each other's messages. Production keeps the plain
+    # name for continuity.
+    env = settings.environment
+    env_segment = "" if env == "production" else f"{env}-"
     queues = []
     for exchange in EXCHANGES:
         suffix = exchange.rsplit("/", 1)[-1]
         queues.append(
             Queue(
-                name=f"queue/{user}/build-repair-{suffix}",
+                name=f"queue/{user}/build-repair-{env_segment}{suffix}",
                 exchange=Exchange(exchange, type="topic", no_declare=True),
                 routing_key="#",
                 durable=True,

services/hackbot-pulse-listener/app/regression.py

@@ -0,0 +1,165 @@
+import logging
+import time
+
+from mozci.errors import ParentPushNotFound
+from mozci.push import Push
+
+logger = logging.getLogger(__name__)
+
+# Maximum number of ancestor pushes to walk back over coalescing gaps before
+# giving up. Mirrors mozci's own MAX_DEPTH.
+MAX_DEPTH = 20
+
+# When the nearest ancestor that ran the build has not produced a decisive
+# result yet, wait for it in-process instead of racing ahead and misreporting an
+# inherited failure as new. A build can take tens of minutes and may hit an
+# infra exception and be auto-retried, so we poll until it settles or the
+# deadline elapses (well above a normal build + one retry). Fails open after the
+# deadline so a real regression is never silently dropped.
+POLL_INTERVAL_SECONDS = 120
+MAX_WAIT_SECONDS = 60 * 60
+
+# A green build; mozci reports it as "passed" (Taskcluster) or "success"
+# (Treeherder). Failures are read from mozci's Task.failed attribute instead of
+# reimplementing the vocabulary.
+_PASSED_RESULTS = ("passed", "success")
+
+# A build in one of these states, or with one of these (infra) results, has not
+# settled: it is still running or was retried after an exception, so its outcome
+# is not knowable yet and we wait for it. Anything else that is not a decisive
+# pass/fail (unscheduled, canceled, superseded, ...) or a build that never ran
+# at all (coalesced) is treated as non-decisive and skipped.
+_UNSETTLED_STATES = ("pending", "running", "exception")
+_UNSETTLED_RESULTS = ("exception", "retry")
+
+# Sentinel meaning the decision cannot be made yet because an ancestor build has
+# not settled; the caller waits and re-checks.
+_PENDING = object()
+
+
+def _build_status(push: Push, label: str):
+    """Return 'passed', 'failed', _PENDING, or None for a build label on a push.
+
+    'passed'/'failed' come from a run with a decisive result. _PENDING means the
+    result is not knowable yet: a task exists but has not settled (still running,
+    or exceptioned and awaiting a retry), or no task is visible yet but the build
+    was scheduled to run on this push (its result has not propagated). None means
+    the build produced no decisive result and none is coming: it was coalesced /
+    never scheduled, or it only reached a non-decisive terminal state. None is
+    deliberately non-decisive so coalescing gaps are skipped and never suppress a
+    real regression. Retriggers are collapsed: any green run means 'passed'; only
+    a genuine build failure counts as 'failed'.
+    """
+    label_tasks = [t for t in push.tasks if t.label == label]
+    if label_tasks:
+        if any(t.result in _PASSED_RESULTS for t in label_tasks):
+            return "passed"
+        # mozci's Task.failed also counts `exception` as a failure, but we treat
+        # an exceptioned build as unsettled (infra, likely auto-retried) and wait
+        # for the retry rather than declaring the parent failed.
+        if any(t.failed and t.result not in _UNSETTLED_RESULTS for t in label_tasks):
+            return "failed"
+        if any(
+            t.state in _UNSETTLED_STATES or t.result in _UNSETTLED_RESULTS
+            for t in label_tasks
+        ):
+            return _PENDING
+        # Ran but reached a non-decisive terminal state (canceled, ...): skip.
+        return None
+
+    # No task for this label on the push. Distinguish a build that was scheduled
+    # to run here but whose result is not visible yet (wait) from one that was
+    # never scheduled / coalesced away (skip), using the decision task's
+    # scheduled set (available well before the builds themselves finish).
+    try:
+        scheduled = label in push.scheduled_task_labels
+    except Exception:
+        logger.debug("Could not read scheduled task labels for %s", push.rev)
+        scheduled = False
+    return _PENDING if scheduled else None
+
+
+def _classify(branch: str, rev: str, label: str):
+    """Walk ancestors once. Returns True (new), False (inherited), or _PENDING.
+
+    A fresh Push is built each call so re-checks re-fetch live data (recent
+    pushes are not finalized in mozci, so their tasks are never served from
+    cache).
+    """
+    ancestor = Push(rev, branch=branch)
+    for _ in range(MAX_DEPTH):
+        try:
+            ancestor = ancestor.parent
+        except ParentPushNotFound:
+            break
+        status = _build_status(ancestor, label)
+        if status is None:
+            continue
+        if status is _PENDING:
+            logger.info(
+                "Build %s not settled yet at %s; deferring decision for %s",
+                label,
+                ancestor.rev,
+                rev,
+            )
+            return _PENDING
+        if status == "failed":
+            logger.info(
+                "Build %s already failing at %s; inherited failure at %s",
+                label,
+                ancestor.rev,
+                rev,
+            )
+            return False
+        logger.info(
+            "Build %s passed at %s; new failure introduced at %s",
+            label,
+            ancestor.rev,
+            rev,
+        )
+        return True
+
+    logger.warning(
+        "No ancestor within %s pushes ran build %s; running agent", MAX_DEPTH, label
+    )
+    return True
+
+
+def is_new_build_failure(branch: str, rev: str, label: str) -> bool:
+    """Return True if this push introduced the failure, False if it inherited it.
+
+    Walks back over pushes that did not run the build (coalescing) until it
+    finds the nearest ancestor that did. When that ancestor's build has not
+    settled yet, waits in-process and re-checks until it produces a decisive
+    result or MAX_WAIT_SECONDS elapses, rather than racing ahead and
+    misreporting an inherited failure as new. Fails open (returns True) on any
+    mozci/network error, if the build stays unsettled past the deadline, or if
+    no ancestor within MAX_DEPTH ran the build, so we never silently drop a real
+    regression.
+    """
+    try:
+        deadline = time.monotonic() + MAX_WAIT_SECONDS
+        while True:
+            result = _classify(branch, rev, label)
+            if result is not _PENDING:
+                return result
+            if time.monotonic() >= deadline:
+                break
+            logger.info(
+                "Waiting %ss for an unsettled ancestor build of %s (%s)",
+                POLL_INTERVAL_SECONDS,
+                rev,
+                label,
+            )
+            time.sleep(POLL_INTERVAL_SECONDS)
+    except Exception:
+        logger.exception("Regression check failed for %s@%s; running agent", label, rev)
+        return True
+
+    logger.warning(
+        "Build %s still unsettled after %ss at %s; running agent",
+        label,
+        MAX_WAIT_SECONDS,
+        rev,
+    )
+    return True

services/hackbot-pulse-listener/deploy.sh

@@ -100,6 +100,7 @@ gcloud builds submit "${ROOT_DIR}" \
 
 echo "==> Deploying worker pool"
 ENV_VARS="HACKBOT_API_URL=${HACKBOT_API_URL},HACKBOT_UI_URL=${HACKBOT_UI_URL}"
+ENV_VARS="${ENV_VARS},ENVIRONMENT=production"
 ENV_VARS="${ENV_VARS},PULSE_USER=${PULSE_USER},WATCHED_REPOS=${WATCHED_REPOS}"
 ENV_VARS="${ENV_VARS},NOTIFICATION_SENDER=${NOTIFICATION_SENDER}"
 ENV_VARS="${ENV_VARS},NOTIFICATION_TEAM_EMAIL=${NOTIFICATION_TEAM_EMAIL}"

services/hackbot-pulse-listener/pyproject.toml

@@ -5,13 +5,17 @@ description = "Listens to Taskcluster build failures and triggers the build-repa
 requires-python = ">=3.12"
 dependencies = [
     "kombu>=5.6,<6",
+    "mozci~=2.4.8",
     "taskcluster>=97.1,<100.4",
     "httpx>=0.26.0",
     "pydantic-settings>=2.1.0",
     "sendgrid>=6.12.5",
     "markdown2>=2.4.0",
     "cachetools>=5.3.0",
     "sentry-sdk>=2.51.0",
+    # mozci imports zstandard eagerly (default cache serializer) but only
+    # declares it under its optional `cache` extra, so pull it in explicitly.
+    "zstandard~=0.25.0",
 ]
 
 [project.optional-dependencies]