Unblock integration CI: stop leaking task defs, drop exact-404-wording assert

scripts/prune_leaked_test_task_defs.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+"""Delete task definitions left behind by integration-test runs.
+
+The integration suites register per-run task defs (``sync_basic_<run id>``,
+``async_lease_heartbeat_task_<run id>``, ...) and, before the tearDownClass
+cleanup was added, never removed them. On a shared server those accumulate
+until the account hits its Task Definitions cap, at which point every
+registration answers::
+
+    402 System has reached the maximum allowed Task Definitions limit of 1000.
+
+and the integration jobs fail on unrelated branches.
+
+The suites now prune stale leftovers themselves at session start (see
+tests/integration/leaked_task_defs.py), so this script is for pruning by hand
+— including the defs too recent for the automatic pass to touch.
+
+Dry run by default: it prints what it would delete and exits. Pass --delete to
+actually remove them. Reads the usual CONDUCTOR_SERVER_URL /
+CONDUCTOR_AUTH_KEY / CONDUCTOR_AUTH_SECRET environment.
+
+    python scripts/prune_leaked_test_task_defs.py             # list matches
+    python scripts/prune_leaked_test_task_defs.py --delete    # remove them
+"""
+
+import argparse
+import sys
+
+from conductor.client.configuration.configuration import Configuration
+from conductor.client.orkes.orkes_metadata_client import OrkesMetadataClient
+from tests.integration.leaked_task_defs import (
+    STALE_AFTER_SECONDS,
+    is_leaked_task_def,
+    stale_leaked_task_defs,
+)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--delete",
+        action="store_true",
+        help="actually unregister the matches (default: dry run)",
+    )
+    parser.add_argument(
+        "--include-recent",
+        action="store_true",
+        help=(
+            "also prune defs newer than "
+            f"{STALE_AFTER_SECONDS // 3600}h — only safe when no integration "
+            "run is in flight, since a concurrent run's defs are fair game"
+        ),
+    )
+    args = parser.parse_args()
+
+    config = Configuration()
+    client = OrkesMetadataClient(config)
+
+    all_defs = client.get_all_task_defs()
+    if args.include_recent:
+        leaked = sorted(d.name for d in all_defs if is_leaked_task_def(d.name))
+    else:
+        leaked = stale_leaked_task_defs(all_defs)
+
+    print(f"server:          {config.host}")
+    print(f"task defs total: {len(all_defs)}")
+    print(f"test leftovers:  {len(leaked)}")
+
+    if not leaked:
+        return 0
+
+    if not args.delete:
+        for name in leaked:
+            print(f"  would delete {name}")
+        print("\nDry run — re-run with --delete to remove these.")
+        return 0
+
+    failed = 0
+    for name in leaked:
+        try:
+            client.unregister_task_def(name)
+            print(f"  deleted {name}")
+        except Exception as e:
+            failed += 1
+            print(f"  FAILED  {name}: {e}", file=sys.stderr)
+
+    print(f"\ndeleted {len(leaked) - failed} of {len(leaked)}")
+    return 1 if failed else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/integration/client/orkes/test_orkes_clients.py

@@ -579,7 +579,10 @@ def __test_workflow_execution_lifecycle(self):
             workflow = self.workflow_client.get_workflow(workflow_uuid, False)
         except ApiException as e:
             assert e.code == 404
-            assert str(e.message).lower() == "workflow with id: {} not found.".format(workflow_uuid)
+            # The server's wording for this 404 changes across releases
+            # ("Workflow with Id: X not found." -> "No execution found for id: X"),
+            # so assert on the durable part: it is a 404 naming this execution.
+            assert workflow_uuid in str(e.message)
 
     def __test_task_execution_lifecycle(self):
 

tests/integration/conftest.py

@@ -52,6 +52,36 @@ def _check_server_connectivity():
     return _server_available, _skip_reason
 
 
+_reclaimed = False
+
+
+def _reclaim_once():
+    """Reclaim task-def quota leaked by earlier runs, before any test registers.
+
+    Without this a filled-up account stays filled: registration answers 402 for
+    every branch, so no run gets far enough to clean up after itself.
+
+    Called from both entry points below — pytest_sessionstart covers the CI
+    invocations (tests/integration/conftest.py is an initial conftest for every
+    bucket), skip_if_server_unavailable covers a session that reaches the
+    unittest suites some other way. The flag makes the second call a no-op.
+    """
+    global _reclaimed
+    if _reclaimed:
+        return
+    _reclaimed = True
+
+    available, _ = _check_server_connectivity()
+    if not available:
+        return
+    from tests.integration.leaked_task_defs import reclaim_task_def_quota
+    reclaim_task_def_quota(Configuration())
+
+
+def pytest_sessionstart(session):
+    _reclaim_once()
+
+
 def skip_if_server_unavailable():
     """
     Call from unittest.TestCase.setUpClass to skip the entire test class
@@ -70,6 +100,61 @@ def setUpClass(cls):
         raise unittest.SkipTest(reason)
 
 
+# ---------------------------------------------------------------------------
+# Metadata cleanup
+# ---------------------------------------------------------------------------
+
+def cleanup_metadata(config, task_defs=(), workflow_defs=()):
+    """Best-effort delete of metadata a test class registered.
+
+    Suites that register per-run task defs (``worker_task(...,
+    register_task_def=True)`` with a RUN_ID-suffixed name) used to leave every
+    one behind, so each CI run added a handful to the shared server until it
+    hit its Task Definitions cap and answered 402 to all further
+    registrations. Call this from tearDownClass.
+
+    Failures are logged and swallowed: cleanup must never turn a passing suite
+    red, and a def that was never registered is nothing to report.
+
+    ``workflow_defs`` items are names (version 1 assumed) or (name, version).
+    """
+    from conductor.client.orkes.orkes_metadata_client import OrkesMetadataClient
+
+    client = OrkesMetadataClient(config)
+
+    def _log(what, e):
+        # "no such definition" is the normal case for anything a deselected or
+        # skipped test never registered, so it logs at debug — only a cleanup
+        # that failed for some other reason is worth a warning.
+        level = logging.DEBUG if _is_missing(e) else logging.WARNING
+        logger.log(level, "cleanup: could not unregister %s: %s", what, e)
+
+    for name in task_defs:
+        try:
+            client.unregister_task_def(name)
+        except Exception as e:
+            _log(f"task def {name}", e)
+
+    for entry in workflow_defs:
+        name, version = entry if isinstance(entry, tuple) else (entry, 1)
+        try:
+            client.unregister_workflow_def(name, version)
+        except Exception as e:
+            _log(f"workflow def {name} v{version}", e)
+
+
+def _is_missing(exc):
+    """True when a delete failed only because the definition was not there.
+
+    The server reports this inconsistently — 404, or a 500 whose body reads
+    "No such task definition" / "No such workflow definition" — so match on
+    both the status and the text.
+    """
+    if getattr(exc, "code", None) == 404:
+        return True
+    return "no such" in str(getattr(exc, "body", "") or exc).lower()
+
+
 # ---------------------------------------------------------------------------
 # Pytest session-scoped fixtures
 # ---------------------------------------------------------------------------

tests/integration/leaked_task_defs.py

@@ -0,0 +1,112 @@
+"""Recognise and reclaim task definitions left behind by integration runs.
+
+The integration suites register per-run task defs
+(``sync_basic_<run id>``, ``async_lease_heartbeat_task_<run id>``, ...). Until
+the ``tearDownClass`` cleanup landed, every run left its own behind, and on the
+shared server they accumulated until the account hit its cap::
+
+    402 System has reached the maximum allowed Task Definitions limit of 1000.
+
+At that point registration fails for every branch, so unrelated PRs go red.
+Cleanup on teardown stops the leak; this module reclaims what earlier runs
+already leaked, and is also what ``scripts/prune_leaked_test_task_defs.py``
+matches on.
+"""
+
+import logging
+import os
+import re
+import time
+
+logger = logging.getLogger(__name__)
+
+# Prefixes owned by the integration suites, each followed by a per-run id.
+# A name is only a candidate if it matches one of these AND ends in a run id,
+# so a hand-registered or production task def is never touched.
+TEST_PREFIXES = (
+    # tests/integration/test_comprehensive_e2e.py
+    "sync_basic_",
+    "async_basic_",
+    "complex_schema_",
+    "task_in_progress_",
+    "failing_task_",
+    # tests/integration/test_lease_extension.py
+    "lease_heartbeat_task_",
+    "lease_no_heartbeat_task_",
+    # tests/integration/test_async_lease_extension.py
+    "async_lease_heartbeat_task_",
+    "async_lease_no_heartbeat_task_",
+    "async_lease_fast_with_hb_",
+    "async_lease_fast_no_hb_",
+    # tests/integration/client/orkes/test_orkes_clients.py (shortuuid suffix)
+    "IntegrationTestOrkesClientsTask_",
+)
+
+# uuid4().hex[:8] for most suites; shortuuid for test_orkes_clients.
+_RUN_ID = re.compile(r"^(?:[0-9a-f]{8}|[0-9A-Za-z]{20,25})$")
+
+# Four integration jobs run in parallel, and other PRs run at the same time.
+# Only defs older than this are reclaimed, so a concurrent run never has the
+# task def it is mid-test on deleted from under it.
+STALE_AFTER_SECONDS = 2 * 60 * 60
+
+
+def is_leaked_task_def(name):
+    """True if ``name`` is a per-run task def owned by the integration suites."""
+    for prefix in TEST_PREFIXES:
+        if name.startswith(prefix) and _RUN_ID.match(name[len(prefix):]):
+            return True
+    return False
+
+
+def stale_leaked_task_defs(task_defs, now=None):
+    """Names in ``task_defs`` that are leaked AND older than STALE_AFTER_SECONDS.
+
+    A def with no ``create_time`` is left alone: without an age there is no way
+    to tell it from one a concurrent run just registered.
+    """
+    cutoff_ms = ((now if now is not None else time.time()) - STALE_AFTER_SECONDS) * 1000
+    return sorted(
+        d.name
+        for d in task_defs
+        if is_leaked_task_def(d.name) and (d.create_time or 0) and d.create_time < cutoff_ms
+    )
+
+
+def reclaim_task_def_quota(config):
+    """Delete stale leaked task defs so registration has room again.
+
+    Best-effort and never raises: this runs before the suites do, and a server
+    that will not answer is the tests' problem to report, not this helper's.
+    Returns the number deleted. Set CONDUCTOR_SKIP_TASK_DEF_RECLAIM=1 to skip.
+    """
+    if os.environ.get("CONDUCTOR_SKIP_TASK_DEF_RECLAIM"):
+        return 0
+
+    from conductor.client.orkes.orkes_metadata_client import OrkesMetadataClient
+
+    client = OrkesMetadataClient(config)
+    try:
+        all_defs = client.get_all_task_defs()
+    except Exception as e:
+        logger.warning("reclaim: could not list task defs: %s", e)
+        return 0
+
+    stale = stale_leaked_task_defs(all_defs)
+    if not stale:
+        return 0
+
+    deleted = 0
+    for name in stale:
+        try:
+            client.unregister_task_def(name)
+            deleted += 1
+        except Exception as e:
+            logger.warning("reclaim: could not unregister %s: %s", name, e)
+
+    # print(), not logger: logging is not configured yet at session start.
+    print(
+        f"reclaimed {deleted} of {len(stale)} stale test task defs "
+        f"({len(all_defs)} defs on {config.host} before pruning)"
+    )
+    return deleted

tests/integration/test_async_lease_extension.py

@@ -232,6 +232,17 @@ def tearDownClass(cls):
         handler = getattr(cls, '_task_handler', None)
         if handler is not None:
             handler.stop_processes()
+        from tests.integration.conftest import cleanup_metadata
+        cleanup_metadata(
+            cls.config,
+            task_defs=(HEARTBEAT_TASK, NO_HEARTBEAT_TASK, FAST_HB_TASK, FAST_NO_HB_TASK),
+            workflow_defs=(
+                f'test_async_lease_heartbeat_{RUN_ID}',
+                f'test_async_lease_no_heartbeat_{RUN_ID}',
+                f'test_async_perf_with_hb_{RUN_ID}',
+                f'test_async_perf_no_hb_{RUN_ID}',
+            ),
+        )
 
     def _register_workflow(self, wf_name, task_names):
         """Register a workflow with one or more tasks in sequence."""

tests/integration/test_comprehensive_e2e.py

@@ -586,6 +586,12 @@ def tearDownClass(cls):
         if os.path.exists(cls.metrics_dir):
             import shutil
             shutil.rmtree(cls.metrics_dir)
+        from tests.integration.conftest import cleanup_metadata
+        cleanup_metadata(
+            cls.config,
+            task_defs=cls.EXPECTED_WORKERS,
+            workflow_defs=(WF_NAME,),
+        )
         print("\n✓ Cleanup complete")
 
 

tests/integration/test_lease_extension.py

@@ -146,6 +146,18 @@ def setUpClass(cls):
         cls.metadata_client = OrkesMetadataClient(cls.config)
         cls.workflow_client = OrkesWorkflowClient(cls.config)
 
+    @classmethod
+    def tearDownClass(cls):
+        from tests.integration.conftest import cleanup_metadata
+        cleanup_metadata(
+            cls.config,
+            task_defs=(HEARTBEAT_TASK, NO_HEARTBEAT_TASK),
+            workflow_defs=(
+                f'test_lease_heartbeat_{RUN_ID}',
+                f'test_lease_no_heartbeat_{RUN_ID}',
+            ),
+        )
+
     def _register_workflow(self, wf_name, task_name):
         """Register a single-task workflow."""
         workflow = WorkflowDef(name=wf_name, version=1)