From dff986e506ba3e27a27510f96e2284b6d904e7f6 Mon Sep 17 00:00:00 2001 From: Stephen Gallagher Date: Fri, 31 Jul 2026 09:34:02 -0400 Subject: [PATCH 1/2] Cancel tagging when repo times out When Koji is very busy, sometimes the initial side-tag creation will fail because a repo cannot be regenerated within the timeout. In this situation, we were deleting the side-tag and retrying, but we didn't actually cancel all of the pending tag tasks for that side-tag buildroot. In situations where the Koji queue is extremely backlogged, this would only serve to make that backlog worse. For maintainability, this converts side-tag creation into a new Python class. Signed-off-by: Stephen Gallagher --- elnbuildsync/kojihelpers/builds.py | 31 ++++++ elnbuildsync/kojihelpers/tags.py | 150 +++++++++++++++++++++-------- elnbuildsync/rebuildbatch.py | 33 ++++--- 3 files changed, 160 insertions(+), 54 deletions(-) diff --git a/elnbuildsync/kojihelpers/builds.py b/elnbuildsync/kojihelpers/builds.py index e29e583..33ff7cb 100644 --- a/elnbuildsync/kojihelpers/builds.py +++ b/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 diff --git a/elnbuildsync/kojihelpers/tags.py b/elnbuildsync/kojihelpers/tags.py index 4a12f8c..ca03b1f 100644 --- a/elnbuildsync/kojihelpers/tags.py +++ b/elnbuildsync/kojihelpers/tags.py @@ -16,11 +16,14 @@ # 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 @@ -28,52 +31,123 @@ 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 = {} diff --git a/elnbuildsync/rebuildbatch.py b/elnbuildsync/rebuildbatch.py index 2c6bc42..d37a3bd 100644 --- a/elnbuildsync/rebuildbatch.py +++ b/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,8 +92,8 @@ 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 @@ -100,13 +101,13 @@ async def _create_and_populate_side_tag( 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 From 739f8d57e9fc8b065f3d01d78fe1b4671291af8d Mon Sep 17 00:00:00 2001 From: Stephen Gallagher Date: Mon, 3 Aug 2026 13:01:19 -0400 Subject: [PATCH 2/2] Normalize the errors returned by wait_for_nvrs_in_tag Signed-off-by: Stephen Gallagher --- elnbuildsync/kojihelpers/tags.py | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/elnbuildsync/kojihelpers/tags.py b/elnbuildsync/kojihelpers/tags.py index ca03b1f..044b322 100644 --- a/elnbuildsync/kojihelpers/tags.py +++ b/elnbuildsync/kojihelpers/tags.py @@ -237,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 @@ -250,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):