Cancel tagging when repo times out

elnbuildsync/kojihelpers/builds.py

@@ -151,6 +151,37 @@ async def cancel_task(task_id):
         logger.exception("Could not cancel task %s. Ignoring.", task_id)
 
 
+async def cancel_tasks(task_ids: list[int]) -> dict[int, dict]:
+    """
+    Cancel multiple tasks.
+
+    :param task_ids: List of task IDs to cancel
+    :returns: A dictionary mapping task_id -> result dict
+    """
+    logger.debug(f"Canceling {len(task_ids)} tasks")
+    results: dict[int, dict] = {}
+    try:
+        results = await call_koji(_cancel_multiple_tasks_thread, task_ids, recurse=True)
+    except Exception:
+        # Cancellation is best-effort
+        logger.exception("Could not cancel tasks %s. Ignoring.", task_ids)
+
+    return results
+
+
+def _cancel_multiple_tasks_thread(bsys, task_ids, **kwargs):
+    task_vcalls = {}
+    with bsys.multicall(batch=config.koji_batch) as mc:
+        for task_id in task_ids:
+            task_vcalls[task_id] = mc.cancelTask(task_id, **kwargs)
+
+    results = {}
+    for task_id, vcall in task_vcalls.items():
+        results[task_id] = vcall.result
+
+    return results
+
+
 async def promote_builds(draft_build_ids):
     promoted_nvrs = await call_koji(_promote_builds_thread, draft_build_ids)
     return promoted_nvrs

elnbuildsync/kojihelpers/tags.py

@@ -16,64 +16,138 @@
 
 # SPDX-License-Identifier: 	GPL-3.0-or-later
 
+from __future__ import annotations
+
 import logging
 
 from cachetools import LRUCache, cached
 from cachetools.keys import hashkey
 from twisted.internet.defer import DeferredList
+from twisted.internet.defer import TimeoutError as DeferredTimeoutError
 
 from .. import config, kojihelpers
 from .connection import call_koji
 
 logger = logging.getLogger(__name__)
 
 
-async def prepare_side_tag(base_tag, initial_build_ids=None):
+class SideTagError(Exception):
+    """
+    A base class for errors related to side tags.
     """
-    Creates a Koji side tag based on @base_tag
 
-    Requests the side-tag and awaits until the repo has been generated.
 
-    :params str base_tag: The build tag to inherit from (e.g. f39-build)
-    :params list initial_packages: The set of build_ids that will be tagged
-    into this side-tag.
-    :return twisted.Deferred: A Twisted Deferred whose callback will fire once
-    the repo is ready for use.
+class SideTagTimeoutError(SideTagError):
+    """
+    A timeout error related to side tags.
     """
 
-    if initial_build_ids is None:
-        initial_build_ids = []
-    # Trigger the creation of the side-tag
-    logger.info(f"Creating side tag from {base_tag}")
-    side_tag_info = await call_koji("createSideTag", base_tag)
-    side_tag_name = side_tag_info["name"]
 
-    logger.debug(f"Side {side_tag_name} created.")
+class SideTag:
+    """
+    A class representing a Koji side tag.
+    """
 
-    if initial_build_ids:
-        # Convert builds to nvrs to make logging easier
-        buildinfos = await kojihelpers.builds.get_multi_buildinfo(initial_build_ids)
-        nvrs = [buildinfo["nvr"] for buildinfo in buildinfos.values()]
+    name: str = None
+
+    def __init__(self, base_tag: str, initial_build_ids: list[int | str] | None = None):
+        """
+        Initialize the SideTag object.
+
+        :params list initial_build_ids: The list of build IDs to tag into the
+        side tag during creation.
+        """
+        self.base_tag = base_tag
+        self.initial_build_ids = initial_build_ids
+
+        # The name of the side-tag is generated by Koji and will be set later
+        self.name = None
+
+        # Maps build ID/NVR -> tagging task ID (as returned by tag_builds()).
+        # Populated during side-tag preparation and cleared once builds are
+        # tagged. cancel_tasks() is given the task ID values on failure.
+        self._tagging_tasks_index = {}
+
+    @classmethod
+    async def create(
+        cls, base_tag: str, initial_build_ids: list[int | str] | None = None
+    ) -> SideTag:
+        """
+        Create a new SideTag object.
+
+        :params str base_tag: The build tag to inherit from (e.g. f39-build)
+        :params list initial_build_ids: The list of build IDs to tag into the
+        side tag during creation.
+        """
+        side_tag = cls(base_tag, initial_build_ids)
+        await side_tag._prepare()
+        return side_tag
+
+    async def _prepare(self):
+        """
+        Prepare the SideTag object for use. It creates the side tag in Koji,
+        tags the initial builds into it, and waits for them to appear in the
+        tag.
+        """
+
+        build_ids_to_tag = []
+        if self.initial_build_ids is not None:
+            build_ids_to_tag.extend(self.initial_build_ids)
+
+        # Trigger the creation of the side-tag
+        logger.info(f"Creating side tag from {self.base_tag}")
+        side_tag_info = await call_koji("createSideTag", self.base_tag)
+        self.name = side_tag_info["name"]
+
+        logger.debug(f"Side {self.name} created.")
+
+        if build_ids_to_tag:
+            # Convert builds to nvrs to make logging easier
+            buildinfos = await kojihelpers.builds.get_multi_buildinfo(build_ids_to_tag)
+            nvrs = [buildinfo["nvr"] for buildinfo in buildinfos.values()]
+
+            # Tag the builds
+            self._tagging_tasks_index = await tag_builds(self.name, nvrs)
+
+            # Wait for the builds to appear in the tag
+            results = await wait_for_nvrs_in_tag(self.name, nvrs)
+
+            if any(not success for success, _ in results):
+                # If any of the builds failed to tag, cancel the tagging tasks,
+                # remove the partial side-tag, and raise from the first exception.
+
+                exc = next((val for success, val in results if not success), None)
+
+                # Cancel all of the tagging tasks. Even if they completed
+                # already, this should be safe to do, and we want to be sure.
+                await kojihelpers.builds.cancel_tasks(
+                    list(self._tagging_tasks_index.values())
+                )
 
-        # Tag the builds
-        # We'll ignore the task index here, since we're actually going to
-        # monitor the tag, rather than the tasks.
-        _ = await tag_builds(side_tag_name, nvrs)
+                # Drop the partially created side-tag before propagating.
+                await self.remove()
 
-        # Wait for the builds to appear in the tag
-        results = await wait_for_nvrs_in_tag(side_tag_name, nvrs)
-        for success, value in results:
-            if success:
-                logger.info(f"Build {value} tagged into {side_tag_name}")
-            else:
-                # The most likely scenario here is that the tagging timed out,
-                # so we'll just proceed. Failures here are not really
-                # recoverable. Log and continue.
-                logger.error(
-                    f"Build failed to tag into {side_tag_name}", exc_info=value
-                )
+                # Check if the exception is a TimeoutError
+                if isinstance(exc, DeferredTimeoutError):
+                    raise SideTagTimeoutError(
+                        "Failed to tag builds into side tag"
+                    ) from exc
+                else:
+                    raise SideTagError("Failed to tag builds into side tag") from exc
+
+            # All the builds were tagged successfully, so log them
+            for _, value in results:
+                logger.info(f"Build {value} tagged into {self.name}")
+
+            # Clear the tagging tasks index
+            self._tagging_tasks_index = {}
 
-    return side_tag_name
+    async def remove(self):
+        """
+        Remove the side tag.
+        """
+        await call_koji("removeSideTag", self.name)
+        logger.info(f"Removed side tag {self.name}")
 
 
 async def tag_builds(tag, build_ids):
@@ -82,7 +156,7 @@ async def tag_builds(tag, build_ids):
 
     :params str tag: The tag name to tag into
     :params list build_ids: The list of nvrs or build IDs to tag
-    :return dict: A dictionary of task_id -> Koji vcall
+    :return dict: A dictionary of build_id/NVR -> tagging task_id
     """
     task_index = await call_koji(_tag_builds_thread, tag, build_ids)
     logger.debug(f"Tagged {len(build_ids)} builds into {tag}")
@@ -96,7 +170,7 @@ def _tag_builds_thread(bsys, tag, build_ids):
 
     :params str tag: The tag name to tag into
     :params list build_ids: The list of nvrs or build IDs to tag
-    :return dict: A dictionary of task_id -> Koji vcall
+    :return dict: A dictionary of build_id/NVR -> tagging task_id
     """
     build_vcalls = {}
 
@@ -163,7 +237,9 @@ async def wait_for_nvrs_in_tag(tag, nvrs):
 
     :params str tag: The tag name to wait for
     :params list nvrs: The list of nvrs to wait for
-    :return list: A list of results
+    :return list: A list of (success, value) results. On failure, ``value`` is
+        the underlying exception (e.g. DeferredTimeoutError), not a Twisted
+        Failure, so callers can isinstance-check timeout errors.
     """
     # Imported lazily to avoid a circular import with listener/batching.
     from .. import listener
@@ -176,7 +252,16 @@ async def wait_for_nvrs_in_tag(tag, nvrs):
         deferreds.append(deferred)
 
     result = await DeferredList(deferreds, consumeErrors=True)
-    return result
+    # DeferredList(consumeErrors=True) wraps failures as Failure objects.
+    # Unwrap to the underlying exception so timeout handling can match
+    # DeferredTimeoutError directly.
+    unwrapped = []
+    for success, value in result:
+        if success:
+            unwrapped.append((True, value))
+        else:
+            unwrapped.append((False, value.value if hasattr(value, "value") else value))
+    return unwrapped
 
 
 async def get_nvrs_from_tag(tag):

elnbuildsync/rebuildbatch.py

@@ -17,6 +17,8 @@
 # SPDX-License-Identifier: 	GPL-3.0-or-later
 
 
+from __future__ import annotations
+
 import logging
 import os
 from collections import defaultdict
@@ -26,7 +28,6 @@
 
 from bodhi.client.bindings import BodhiClient, BodhiClientException
 from tenacity import retry, stop_after_delay, wait_exponential
-from twisted.internet.defer import TimeoutError as DeferredTimeoutError
 from twisted.internet.threads import deferToThread
 
 from . import config, kojihelpers
@@ -91,22 +92,22 @@ async def async_init(self):
         return self
 
     async def _create_and_populate_side_tag(
-        self, build_ids: list[int], promote_builds: bool = False
-    ) -> tuple[str, list[str]]:
+        self, build_ids: list[int | str], promote_builds: bool = False
+    ) -> tuple[kojihelpers.tags.SideTag, list[int | str]]:
         """
         Creates a side-tag for this batch. If promote_builds is True, the builds
         will be promoted before tagging. This will return the NVRs that were
         tagged, which may not be the same as the input build_ids/NVRs if some
         builds were not able to be promoted (e.g. if another draft build was
         already promoted with the same NVR).
 
-        :param build_ids: The list of build_ids to tag into the side-tag.
-        :type build_ids: list[int]
+        :param build_ids: Build IDs or NVR strings to tag into the side-tag.
+        :type build_ids: list[int | str]
         :param promote_builds: Whether to promote draft builds before tagging.
 
-        :return: The side-tag name and the refs that were tagged (promoted NVRs
+        :return: The SideTag object and the refs that were tagged (promoted NVRs
             when promote_builds is True, otherwise the input build_ids/NVRs).
-        :rtype: tuple[str, list[str]]
+        :rtype: tuple[kojihelpers.tags.SideTag, list[int | str]]
         """
 
         if promote_builds:
@@ -120,11 +121,11 @@ async def _create_and_populate_side_tag(
 
         while True:
             try:
-                side_tag = await kojihelpers.tags.prepare_side_tag(
+                side_tag = await kojihelpers.tags.SideTag.create(
                     self._side_tag_base,
                     build_nvrs,
                 )
-            except DeferredTimeoutError:
+            except kojihelpers.tags.SideTagTimeoutError:
                 # Keep retrying to create a side-tag.
                 # Any other exception will be propagated up the stack.
                 logger.warning(
@@ -246,8 +247,8 @@ async def run(self):
         # Remove the side-tag where we performed the rebuilds.
         # The update tag will be automatically removed when the Bodhi update
         # makes it to stable.
-        logger.info(f"Removing side-tag {self.side_tag}")
-        await kojihelpers.tags.remove_side_tag(self.side_tag)
+        logger.info(f"Removing side-tag {self.side_tag.name}")
+        await self.side_tag.remove()
 
     async def _create_and_submit_bodhi_updates(
         self, build_nvrs: list[str]
@@ -282,11 +283,11 @@ async def _process_batch(batch_nvrs: list[str]) -> None:
                 )
                 return
 
-            logger.info(f"Submitting Bodhi update for {update_tag}")
+            logger.info(f"Submitting Bodhi update for {update_tag.name}")
             try:
                 await deferToThread(self._submit_bodhi_update, update_tag)
             except Exception:
-                logger.exception(f"Failed to submit Bodhi update for {update_tag}")
+                logger.exception(f"Failed to submit Bodhi update for {update_tag.name}")
                 raise
             logger.debug(f"Submitted Bodhi update for {batch_nvrs}")
             promoted_nvrs.extend(batch_promoted_nvrs)
@@ -304,7 +305,7 @@ async def _process_batch(batch_nvrs: list[str]) -> None:
         stop=stop_after_delay(900),
         reraise=True,
     )
-    def _submit_bodhi_update(self, update_tag: str) -> None:
+    def _submit_bodhi_update(self, update_tag: kojihelpers.tags.SideTag) -> None:
         try:
             # Submitting a Bodhi update is infrequent-enough that it doesn't
             # really make sense to try to cache the connection. Just
@@ -328,10 +329,10 @@ def _submit_bodhi_update(self, update_tag: str) -> None:
             # Rawhide updates.
             bodhi.save(
                 type="unspecified",
-                from_tag=update_tag,
+                from_tag=update_tag.name,
                 notes="Automatic update for ELN rebuild batch",
             )
-            logger.info(f"Submitted Bodhi update for {update_tag}")
+            logger.info(f"Submitted Bodhi update for {update_tag.name}")
         except BodhiClientException as e:
             logger.error(f"Failed to submit Bodhi update: {e}")
             raise