From 19f34698e55bc751412e86e45c3503dc4e4cc261 Mon Sep 17 00:00:00 2001 From: Manan Bhatt Date: Thu, 13 Aug 2026 18:58:05 +0530 Subject: [PATCH 1/8] Assert 404 status, not server prose, for deleted-workflow lookup test_all asserted the exact 404 body for a deleted workflow: assert str(e.message).lower() == "workflow with id: {} not found." The sdkdev v5 deployment was rolled between Aug 10 and Aug 12 and now answers "No execution found for id: ". No SDK or test code changed across that boundary -- integration-test (test-all) was green on Aug 10 08:01 and red from Aug 12 19:35, and rest.py surfaces the server's `message` field verbatim -- so the assert broke on a server-side prose change alone. It has kept CI red on main and on every open PR since. Assert the status code, which is the actual contract, and only that the id appears in the message. Also closes a vacuous pass in the same block: the try had no else or fail, so a get_workflow that did not raise satisfied the test silently. Raise instead if the deleted workflow is still readable -- AssertionError is not an ApiException, so it propagates past the handler. --- tests/integration/client/orkes/test_orkes_clients.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/tests/integration/client/orkes/test_orkes_clients.py b/tests/integration/client/orkes/test_orkes_clients.py index 2811d492..e4ea9ed1 100644 --- a/tests/integration/client/orkes/test_orkes_clients.py +++ b/tests/integration/client/orkes/test_orkes_clients.py @@ -575,11 +575,17 @@ def __test_workflow_execution_lifecycle(self): assert workflow.status == "RUNNING" self.workflow_client.delete_workflow(workflow_uuid) + # Assert on the 404 status, not the server's prose. The message wording is + # not part of the API contract and has changed server-side before, which + # turned CI red without any SDK change. try: - workflow = self.workflow_client.get_workflow(workflow_uuid, False) + self.workflow_client.get_workflow(workflow_uuid, False) + raise AssertionError( + "expected a 404 for deleted workflow {}".format(workflow_uuid) + ) except ApiException as e: assert e.code == 404 - assert str(e.message).lower() == "workflow with id: {} not found.".format(workflow_uuid) + assert workflow_uuid in str(e.message) def __test_task_execution_lifecycle(self): From e59d73f4a3c1e37de3f6681505723a383761da41 Mon Sep 17 00:00:00 2001 From: Manan Bhatt Date: Thu, 13 Aug 2026 19:19:20 +0530 Subject: [PATCH 2/8] Stop asserting server prose; wait out eventually-consistent reads Two failure classes have been turning CI red without any SDK change. 1. Exact-prose 404 asserts. Seven sites compared e.message against the server's exact English sentence. The 5.3.3 -> 5.5.0 upgrade of the shared sdkdev server reworded one of them and broke main. Replaced with _assert_not_found, which asserts the 404 status -- the actual contract -- and only that the identifier appears in the message. It also fails when the resource is still readable after a delete. The bare try/except at each site passed silently in that case, so a delete that did not take effect would not have been caught. 2. Read-after-write on queue size. get_queue_size_for_task was asserted immediately after starting or draining work, but the server enqueues and indexes asynchronously, so the count intermittently came back stale (assert 0 == 2). Wrapped in _await_value, which polls up to 15s for the expected value and returns the last one seen so a real mismatch still reports the actual number. Not touched: __test_workflow_rate_limit has the same read-after-write shape but is dead code -- nothing in run() calls it. --- .../client/orkes/test_orkes_clients.py | 107 +++++++++--------- 1 file changed, 56 insertions(+), 51 deletions(-) diff --git a/tests/integration/client/orkes/test_orkes_clients.py b/tests/integration/client/orkes/test_orkes_clients.py index e4ea9ed1..42497261 100644 --- a/tests/integration/client/orkes/test_orkes_clients.py +++ b/tests/integration/client/orkes/test_orkes_clients.py @@ -53,6 +53,36 @@ def _retry_on_404(func, *args, retries=5, **kwargs): raise +def _assert_not_found(fetch, *identifiers): + # Assert the 404 status, not the server's prose. The wording is not part of + # the API contract: a server release changed "Workflow with id: X not found." + # to "No execution found for id: X" and turned CI red with no SDK change. + # Also fail loudly when the resource is still readable -- the bare + # try/except this replaces passed silently in exactly that case. + try: + fetch() + except ApiException as e: + assert e.code == 404, f"expected a 404, got {e.code}: {e.message}" + for identifier in identifiers: + assert str(identifier) in str(e.message), \ + f"expected {identifier!r} in the 404 message, got: {e.message}" + return + raise AssertionError("expected a 404, but the resource is still readable") + + +def _await_value(fetch, expected, timeout=15, interval=1): + # Queue size is eventually consistent -- the server enqueues and indexes + # asynchronously, so reading straight after starting or draining work + # intermittently returns a stale count. Poll until it settles, then let the + # caller assert on the last value seen so failures still show the mismatch. + deadline = time.time() + timeout + value = fetch() + while value != expected and time.time() < deadline: + time.sleep(interval) + value = fetch() + return value + + class TestOrkesClients: def __init__(self, configuration: Configuration): self.api_client = ApiClient(configuration) @@ -129,11 +159,7 @@ def test_task_lifecycle(self): self.__test_task_execution_lifecycle() self.metadata_client.unregister_task_def(TASK_TYPE) - try: - self.metadata_client.get_task_def(TASK_TYPE) - except ApiException as e: - assert e.code == 404 - assert e.message == "Task {0} not found".format(TASK_TYPE) + _assert_not_found(lambda: self.metadata_client.get_task_def(TASK_TYPE), TASK_TYPE) def test_secret_lifecycle(self): self.secret_client.put_secret(SECRET_NAME, "secret_value") @@ -164,11 +190,7 @@ def test_secret_lifecycle(self): assert self.secret_client.secret_exists(SECRET_NAME) == False self.secret_client.delete_secret(SECRET_NAME + "_2") - - try: - self.secret_client.get_secret(SECRET_NAME + "_2") - except ApiException as e: - assert e.code == 404 + _assert_not_found(lambda: self.secret_client.get_secret(SECRET_NAME + "_2")) def test_scheduler_lifecycle(self, workflowDef): startWorkflowRequest = StartWorkflowRequest( @@ -212,12 +234,9 @@ def test_scheduler_lifecycle(self, workflowDef): assert len(fetched_tags) == 0 self.scheduler_client.delete_schedule(SCHEDULE_NAME) - - try: - schedule = self.scheduler_client.get_schedule(SCHEDULE_NAME) - except ApiException as e: - assert e.code == 404 - assert e.message == "Schedule '{0}' not found".format(SCHEDULE_NAME) + _assert_not_found( + lambda: self.scheduler_client.get_schedule(SCHEDULE_NAME), SCHEDULE_NAME + ) def test_application_lifecycle(self): req = CreateOrUpdateApplicationRequest(APPLICATION_NAME) @@ -263,11 +282,10 @@ def test_application_lifecycle(self): self.authorization_client.delete_access_key(created_app.id, created_access_key.id) self.authorization_client.delete_application(created_app.id) - try: - application = self.authorization_client.get_application(created_app.id) - except ApiException as e: - assert e.code == 404 - assert e.message == "Application '{0}' not found".format(created_app.id) + _assert_not_found( + lambda: self.authorization_client.get_application(created_app.id), + created_app.id + ) def test_user_group_permissions_lifecycle(self, workflowDef): req = UpsertUserRequest("Integration User", ["USER"]) @@ -340,18 +358,10 @@ def test_user_group_permissions_lifecycle(self, workflowDef): self.authorization_client.remove_user_from_group(GROUP_ID, USER_ID) self.authorization_client.delete_user(USER_ID) - try: - self.authorization_client.get_user(USER_ID) - except ApiException as e: - assert e.code == 404 - assert e.message == "User '{0}' not found".format(USER_ID) + _assert_not_found(lambda: self.authorization_client.get_user(USER_ID), USER_ID) self.authorization_client.delete_group(GROUP_ID) - try: - self.authorization_client.get_group(GROUP_ID) - except ApiException as e: - assert e.code == 404 - assert e.message == "Group '{0}' not found".format(GROUP_ID) + _assert_not_found(lambda: self.authorization_client.get_group(GROUP_ID), GROUP_ID) def __test_register_workflow_definition(self, workflowDef: WorkflowDef): self.__create_workflow_definition(workflowDef) @@ -482,12 +492,9 @@ def __poll_workflow_until_complete(self, workflow_id, timeout_seconds=60, def __test_unregister_workflow_definition(self): self.metadata_client.unregister_workflow_def(WORKFLOW_NAME, 1) - - try: - self.metadata_client.get_workflow_def(WORKFLOW_NAME, 1) - except ApiException as e: - assert e.code == 404 - assert e.message == 'No such workflow found by name: {0}, version: 1'.format(WORKFLOW_NAME) + _assert_not_found( + lambda: self.metadata_client.get_workflow_def(WORKFLOW_NAME, 1), WORKFLOW_NAME + ) def __test_task_tags(self): tags = [ @@ -575,17 +582,9 @@ def __test_workflow_execution_lifecycle(self): assert workflow.status == "RUNNING" self.workflow_client.delete_workflow(workflow_uuid) - # Assert on the 404 status, not the server's prose. The message wording is - # not part of the API contract and has changed server-side before, which - # turned CI red without any SDK change. - try: - self.workflow_client.get_workflow(workflow_uuid, False) - raise AssertionError( - "expected a 404 for deleted workflow {}".format(workflow_uuid) - ) - except ApiException as e: - assert e.code == 404 - assert workflow_uuid in str(e.message) + _assert_not_found( + lambda: self.workflow_client.get_workflow(workflow_uuid, False), workflow_uuid + ) def __test_task_execution_lifecycle(self): @@ -612,7 +611,9 @@ def __test_task_execution_lifecycle(self): workflow_uuid_2 = self.workflow_client.start_workflow(startWorkflowRequest) # First task of each workflow is in the queue - assert self.task_client.get_queue_size_for_task(TASK_TYPE) == 2 + assert _await_value( + lambda: self.task_client.get_queue_size_for_task(TASK_TYPE), 2 + ) == 2 polledTask = self.task_client.poll_task(TASK_TYPE) assert polledTask.status == TaskResultStatus.IN_PROGRESS @@ -623,7 +624,9 @@ def __test_task_execution_lifecycle(self): assert taskExecLogs[0].log == "Polled task..." # First task of second workflow is still in the queue - assert self.task_client.get_queue_size_for_task(TASK_TYPE) == 1 + assert _await_value( + lambda: self.task_client.get_queue_size_for_task(TASK_TYPE), 1 + ) == 1 taskResult = TaskResult( workflow_instance_id=polledTask.workflow_instance_id, @@ -672,7 +675,9 @@ def __test_task_execution_lifecycle(self): ) completed += 1 - queue_size = self.task_client.get_queue_size_for_task(TASK_TYPE) + queue_size = _await_value( + lambda: self.task_client.get_queue_size_for_task(TASK_TYPE), 0 + ) print(f'queue size for {TASK_TYPE} is {queue_size}') assert queue_size == 0 From 5f3ba52c633396f21ecede8665391615db75de71 Mon Sep 17 00:00:00 2001 From: Manan Bhatt Date: Thu, 13 Aug 2026 19:48:29 +0530 Subject: [PATCH 3/8] Reclaim leaked task defs; retry 423 contention at scenario level Folds in the remaining deterministic CI failure causes so one PR covers all of them. Task-def quota (402). The integration suites register per-run task defs and never removed them, so on the shared server they accumulated to the 1000 cap and registration then answered 402 for every branch -- which is also what produced the test_05_verify_task_definitions 404s. Adds tearDownClass cleanup to the suites that leak, a conftest session hook that reclaims what earlier runs already left behind, and a manual prune script. Reclaim only touches names matching a known test prefix plus a run id, and only those older than two hours, so a concurrent run never has a def deleted from under it. Salvaged from the closed PR #475. 423 write contention. "Workflow is currently being updated, please retry" is the server asking for a retry, but it is scoped to retry_scenario only, not retry_on_transient: replaying a single non-idempotent call (start_workflow, signal) risks double-executing a write that landed and only contended on the read-back, whereas a scenario re-runs from the top and rebuilds its own state. Service registry 502. That suite called its scenarios bare, so one 502 from the proxy failed the whole run even though is_transient already covers gateway 5xx. Wrapped in retry_scenario with the same shared deadline the other aggregate suites use. Not fixed: test_v2_fallback_intg "workflows did not complete in time". It already retries transients, so this is completion timing, not a transient API error -- possibly real 5.5.0 slowness. Raising its deadline blind would hide that. --- scripts/prune_leaked_test_task_defs.py | 92 ++++++++++++++ .../test_orkes_service_registry_client.py | 29 ++++- tests/integration/conftest.py | 85 +++++++++++++ tests/integration/leaked_task_defs.py | 112 ++++++++++++++++++ tests/integration/retry_helpers.py | 49 ++++++-- .../integration/test_async_lease_extension.py | 11 ++ tests/integration/test_comprehensive_e2e.py | 6 + tests/integration/test_lease_extension.py | 12 ++ 8 files changed, 383 insertions(+), 13 deletions(-) create mode 100644 scripts/prune_leaked_test_task_defs.py create mode 100644 tests/integration/leaked_task_defs.py diff --git a/scripts/prune_leaked_test_task_defs.py b/scripts/prune_leaked_test_task_defs.py new file mode 100644 index 00000000..58d0f0f2 --- /dev/null +++ b/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_``, +``async_lease_heartbeat_task_``, ...) 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()) diff --git a/tests/integration/client/orkes/test_orkes_service_registry_client.py b/tests/integration/client/orkes/test_orkes_service_registry_client.py index f7830a65..ce3cd3ab 100644 --- a/tests/integration/client/orkes/test_orkes_service_registry_client.py +++ b/tests/integration/client/orkes/test_orkes_service_registry_client.py @@ -12,6 +12,10 @@ from conductor.client.http.models.proto_registry_entry import ProtoRegistryEntry from conductor.client.orkes.orkes_service_registry_client import OrkesServiceRegistryClient from conductor.client.http.rest import ApiException +from tests.integration.retry_helpers import ( + DEFAULT_OVERALL_DEADLINE_SECONDS, + retry_scenario, +) SUFFIX = str(uuid()) HTTP_SERVICE_NAME = 'IntegrationTestServiceRegistryHttp_' + SUFFIX @@ -38,11 +42,20 @@ def __init__(self, configuration: Configuration): self.client = OrkesServiceRegistryClient(configuration) logger.info(f'Setting up TestOrkesServiceRegistryClient with config {configuration}') - def run(self) -> None: - """Run all service registry tests""" - self.test_http_service_registry() - self.test_grpc_service() - self.test_proto_operations() + def run(self, deadline=None) -> None: + """Run all service registry tests. + + Each is a scenario: on a transient blip against the shared dev server + (gateway 5xx, 423 contention, status-0 transport hiccup) it retries from + the top until the shared deadline passes. Real failures raise at once. + This suite used to call them bare, so a single 502 from the proxy failed + the whole run. + """ + retry_scenario('test_http_service_registry', self.test_http_service_registry, + deadline=deadline) + retry_scenario('test_grpc_service', self.test_grpc_service, deadline=deadline) + retry_scenario('test_proto_operations', self.test_proto_operations, + deadline=deadline) def setUp(self): """Clean up services before each test""" @@ -276,8 +289,12 @@ def test_all(self): logger.info('START: service registry integration tests') configuration = self.config + # One shared wall-clock budget for the whole suite, as in + # test_workflow_client_intg.test_all. + deadline = time.monotonic() + DEFAULT_OVERALL_DEADLINE_SECONDS + # Run service registry tests - TestOrkesServiceRegistryClient(configuration=configuration).run() + TestOrkesServiceRegistryClient(configuration=configuration).run(deadline=deadline) logger.info('END: service registry integration tests') diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index aede140e..3f7de2a7 100644 --- a/tests/integration/conftest.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/tests/integration/leaked_task_defs.py b/tests/integration/leaked_task_defs.py new file mode 100644 index 00000000..098ca415 --- /dev/null +++ b/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_``, ``async_lease_heartbeat_task_``, ...). 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 diff --git a/tests/integration/retry_helpers.py b/tests/integration/retry_helpers.py index 4f69ed79..3dcc5ba2 100644 --- a/tests/integration/retry_helpers.py +++ b/tests/integration/retry_helpers.py @@ -51,6 +51,18 @@ # handling rather than blind backoff). GATEWAY_STATUSES = (502, 503, 504) +# 423 Locked: "Workflow is currently being updated, please retry". The server +# takes a short-lived lock while applying a write and answers this when a +# concurrent request touches the same workflow. It fires on the shared dev +# server whenever several runs exercise the same workflow at once. +# +# Deliberately NOT part of is_transient: retry_on_transient retries a single +# request, and replaying a non-idempotent call (start_workflow, signal) on a 423 +# risks double-executing a write that actually landed and only contended on the +# read-back. Only retry_scenario honours it, because a scenario re-runs from the +# top and rebuilds its own state. +LOCKED_STATUS = 423 + def is_transient(exc): """True when ``exc`` is a transient blip against the shared dev server @@ -59,6 +71,9 @@ def is_transient(exc): keep-alive — surfaced as status 0/None or the ``transient`` flag) and gateway-class 5xx (502/503/504) returned by the proxy/LB in front of the server (see ``GATEWAY_STATUSES``). + + 423 is excluded here on purpose — see ``LOCKED_STATUS`` and + ``is_scenario_retryable``. """ return isinstance(exc, ApiException) and ( getattr(exc, 'transient', False) @@ -66,10 +81,20 @@ def is_transient(exc): or exc.status in GATEWAY_STATUSES) -def first_transient_api_exception(exc): +def is_scenario_retryable(exc): + """True when ``exc`` is worth re-running a whole scenario for. + + Everything ``is_transient`` covers, plus 423 write contention: safe here + because a scenario re-runs from the top and rebuilds its own state, whereas + replaying one non-idempotent request is not (see ``LOCKED_STATUS``). + """ + return is_transient(exc) or ( + isinstance(exc, ApiException) and exc.status == LOCKED_STATUS) + + +def _first_matching_api_exception(exc, predicate): """Walk the exception chain (``__cause__`` / ``__context__``) and return the - first transient ``ApiException`` (flagged transient, or status 0/None), or - ``None`` if there isn't one. + first ``ApiException`` satisfying ``predicate``, or ``None``. Inner test helpers sometimes catch an ApiException and re-raise it as a bare ``Exception`` (losing the type), so we can't rely on the outermost exception @@ -80,12 +105,22 @@ def first_transient_api_exception(exc): cur = exc while cur is not None and id(cur) not in seen: seen.add(id(cur)) - if is_transient(cur): + if predicate(cur): return cur cur = cur.__cause__ or cur.__context__ return None +def first_transient_api_exception(exc): + """First transient ``ApiException`` in the chain, or ``None``.""" + return _first_matching_api_exception(exc, is_transient) + + +def first_scenario_retryable_api_exception(exc): + """First scenario-retryable ``ApiException`` in the chain, or ``None``.""" + return _first_matching_api_exception(exc, is_scenario_retryable) + + def retry_on_transient(func, *args, retries=4, base_delay=1, **kwargs): """Retry ``func(*args, **kwargs)`` on a transient blip (see ``is_transient``: status 0/None transport hiccups plus gateway-class 502/503/504) with @@ -141,8 +176,8 @@ def retry_scenario(label, func, *args, deadline=None, base_delay=DEFAULT_BASE_DELAY_SECONDS, max_delay=DEFAULT_MAX_DELAY_SECONDS, **kwargs): """Run ``func(*args, **kwargs)``, retrying only on a transient blip (see - ``is_transient``: status 0/None transport hiccups plus gateway-class - 502/503/504) until the shared ``deadline`` passes. + ``is_scenario_retryable``: status 0/None transport hiccups, gateway-class + 502/503/504, and 423 write contention) until the shared ``deadline`` passes. Args: label: Human-readable scenario name for logs. @@ -166,7 +201,7 @@ def retry_scenario(label, func, *args, deadline=None, try: return func(*args, **kwargs) except Exception as e: - transient = first_transient_api_exception(e) + transient = first_scenario_retryable_api_exception(e) if transient is None: raise now = time.monotonic() diff --git a/tests/integration/test_async_lease_extension.py b/tests/integration/test_async_lease_extension.py index d384ac4d..b6829fbe 100644 --- a/tests/integration/test_async_lease_extension.py +++ b/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.""" diff --git a/tests/integration/test_comprehensive_e2e.py b/tests/integration/test_comprehensive_e2e.py index 4fcfac8e..178c09f6 100644 --- a/tests/integration/test_comprehensive_e2e.py +++ b/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") diff --git a/tests/integration/test_lease_extension.py b/tests/integration/test_lease_extension.py index 3b986409..82f1aca4 100644 --- a/tests/integration/test_lease_extension.py +++ b/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) From 9360df447be3dc86dc10dfa9ce86903c08fe3982 Mon Sep 17 00:00:00 2001 From: Manan Bhatt Date: Thu, 13 Aug 2026 22:05:13 +0530 Subject: [PATCH 4/8] Report task state when a workflow fails to complete Both intermittent failures assert only on workflow status, so a red run says "still RUNNING" or "5 pending" and nothing about why. Dump each task's status, domain, pollCount and workerId on the failure path. That separates the possibilities in one run instead of by hypothesis: SCHEDULED means queued but never polled (wrong queue or domain, or no live worker for the type), IN_PROGRESS means polled and leased but never updated, and no tasks at all means the workflow was never decided. Diagnostics only -- no behaviour change. --- tests/integration/test_v2_fallback_intg.py | 25 +++++++++++++++ .../workflow/test_workflow_execution.py | 31 ++++++++++++++++++- 2 files changed, 55 insertions(+), 1 deletion(-) diff --git a/tests/integration/test_v2_fallback_intg.py b/tests/integration/test_v2_fallback_intg.py index b265da79..442e2d0c 100644 --- a/tests/integration/test_v2_fallback_intg.py +++ b/tests/integration/test_v2_fallback_intg.py @@ -171,6 +171,31 @@ def _run_workers(): print(f" Results: {completed} completed, {failed} failed, {len(pending)} pending") + # On failure the workflow status alone ("still RUNNING") says nothing + # about why. Dump the tasks so the next red run distinguishes: + # SCHEDULED -> queued but never polled (wrong queue/domain, or + # no live worker for that task type) + # IN_PROGRESS -> polled and leased, never updated (worker stuck) + # no tasks -> the workflow was never decided (server side) + if pending: + print(f" DIAGNOSTIC: {len(pending)} workflow(s) did not complete") + for wf_id in sorted(pending): + try: + wf = self.workflow_client.get_workflow(wf_id, include_tasks=True) + tasks = wf.tasks or [] + print(f" {wf_id} status={wf.status} tasks={len(tasks)}") + for t in tasks: + print( + f" task={getattr(t, 'task_def_name', '?')} " + f"ref={getattr(t, 'reference_task_name', '?')} " + f"status={getattr(t, 'status', '?')} " + f"domain={getattr(t, 'domain', None)} " + f"pollCount={getattr(t, 'poll_count', None)} " + f"workerId={getattr(t, 'worker_id', None)}" + ) + except Exception as e: + print(f" {wf_id}: could not fetch tasks: {e}") + self.assertEqual(len(pending), 0, f"{len(pending)} workflows did not complete in time") self.assertEqual(completed, workflow_count, f"Expected {workflow_count} completed, got {completed}") diff --git a/tests/integration/workflow/test_workflow_execution.py b/tests/integration/workflow/test_workflow_execution.py index 49e534bf..64aea44b 100644 --- a/tests/integration/workflow/test_workflow_execution.py +++ b/tests/integration/workflow/test_workflow_execution.py @@ -239,6 +239,33 @@ def generate_workflow(workflow_executor: WorkflowExecutor, workflow_name: str = ) +def _describe_tasks(workflow_id: str, workflow_executor: WorkflowExecutor) -> str: + """Task-level detail for a workflow that did not reach COMPLETED. + + "still RUNNING" on its own says nothing about why. The task states separate + the possibilities: SCHEDULED means queued but never polled (wrong + queue/domain, or no live worker for that task type), IN_PROGRESS means + polled and leased but never updated, and no tasks at all means the workflow + was never decided server-side. + """ + try: + wf = workflow_executor.get_workflow(workflow_id=workflow_id, include_tasks=True) + tasks = wf.tasks or [] + if not tasks: + return 'no tasks on the workflow' + return '; '.join( + f"{getattr(t, 'task_def_name', '?')}" + f"[ref={getattr(t, 'reference_task_name', '?')}" + f" status={getattr(t, 'status', '?')}" + f" domain={getattr(t, 'domain', None)}" + f" pollCount={getattr(t, 'poll_count', None)}" + f" workerId={getattr(t, 'worker_id', None)}]" + for t in tasks + ) + except Exception as e: + return f'could not fetch tasks: {e}' + + def validate_workflow_status(workflow_id: str, workflow_executor: WorkflowExecutor) -> None: workflow = workflow_executor.get_workflow( workflow_id=workflow_id, @@ -246,7 +273,9 @@ def validate_workflow_status(workflow_id: str, workflow_executor: WorkflowExecut ) if workflow.status != 'COMPLETED': raise Exception( - f'workflow expected to be COMPLETED, but received {workflow.status}, workflow_id: {workflow_id}' + f'workflow expected to be COMPLETED, but received {workflow.status}, ' + f'workflow_id: {workflow_id}, tasks: ' + f'{_describe_tasks(workflow_id, workflow_executor)}' ) workflow_status = workflow_executor.get_workflow_status( workflow_id=workflow_id, From acca5080642ec78dac8fb6325e49cce9f75ad40a Mon Sep 17 00:00:00 2001 From: Manan Bhatt Date: Thu, 13 Aug 2026 22:22:07 +0530 Subject: [PATCH 5/8] Drop the manual prune script Nothing runs it. The conftest session hook already reclaims stale leaked task defs, which is what actually keeps the account under its cap; the script was a hand-run convenience that came along with the #475 salvage and is not needed for CI to pass. --- scripts/prune_leaked_test_task_defs.py | 92 -------------------------- tests/integration/leaked_task_defs.py | 3 +- 2 files changed, 1 insertion(+), 94 deletions(-) delete mode 100644 scripts/prune_leaked_test_task_defs.py diff --git a/scripts/prune_leaked_test_task_defs.py b/scripts/prune_leaked_test_task_defs.py deleted file mode 100644 index 58d0f0f2..00000000 --- a/scripts/prune_leaked_test_task_defs.py +++ /dev/null @@ -1,92 +0,0 @@ -#!/usr/bin/env python3 -"""Delete task definitions left behind by integration-test runs. - -The integration suites register per-run task defs (``sync_basic_``, -``async_lease_heartbeat_task_``, ...) 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()) diff --git a/tests/integration/leaked_task_defs.py b/tests/integration/leaked_task_defs.py index 098ca415..4270e190 100644 --- a/tests/integration/leaked_task_defs.py +++ b/tests/integration/leaked_task_defs.py @@ -9,8 +9,7 @@ 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. +already leaked. """ import logging From 759dd281c47a700fea4663496a7d7b0eac27dc9a Mon Sep 17 00:00:00 2001 From: Manan Bhatt Date: Thu, 13 Aug 2026 22:27:06 +0530 Subject: [PATCH 6/8] Wait for the blocking task before signalling, instead of sleeping The seven signal scenarios started a workflow, slept 0.5-1.0s, then signalled and asserted on the returned strategy. Every signal strategy describes the task the workflow is parked on, so if the workflow has not got there yet the server answers with no responseType and the assert reads "Expected BLOCKING_TASK, got None". On a loaded shared server one second is not enough, which is why it failed intermittently. Poll for the precondition -- a task in SCHEDULED or IN_PROGRESS -- up to 30s, and give up early if the workflow is already terminal so the caller's own assertion reports the real state. Logs the tasks it saw when the wait times out. --- .../workflow/test_workflow_execution.py | 49 ++++++++++++++++--- 1 file changed, 41 insertions(+), 8 deletions(-) diff --git a/tests/integration/workflow/test_workflow_execution.py b/tests/integration/workflow/test_workflow_execution.py index 64aea44b..87ebe012 100644 --- a/tests/integration/workflow/test_workflow_execution.py +++ b/tests/integration/workflow/test_workflow_execution.py @@ -632,6 +632,39 @@ def _start_complex_workflow(workflow_executor: WorkflowExecutor) -> str: raise +def _wait_for_blocking_task(workflow_executor: WorkflowExecutor, workflow_id: str, + timeout: float = 30.0, interval: float = 0.5): + """Wait until the workflow is actually parked on a task, before signalling. + + Every signal return strategy (BLOCKING_TASK, BLOCKING_WORKFLOW, ...) + describes the task the workflow is currently blocked on. A freshly started + workflow needs a moment to run its first task and schedule that one, and the + fixed sleep(0.5) this replaces was not enough on a loaded shared server: the + signal came back with no responseType at all, surfacing as the intermittent + "Expected BLOCKING_TASK, got None". + + Returns the tasks last seen so callers can report them if the wait times out. + """ + deadline = time.time() + timeout + tasks = [] + while time.time() < deadline: + workflow = workflow_executor.get_workflow(workflow_id=workflow_id, + include_tasks=True) + tasks = workflow.tasks or [] + if any(getattr(t, 'status', None) in ('SCHEDULED', 'IN_PROGRESS') + for t in tasks): + return tasks + if getattr(workflow, 'status', None) not in ('RUNNING', 'PAUSED'): + # Already terminal: nothing is going to block, so stop waiting and + # let the caller's assertion report the real state. + break + time.sleep(interval) + logger.warning( + 'no blocking task on %s after %.0fs; tasks=%s', workflow_id, timeout, + [(getattr(t, 'task_def_name', '?'), getattr(t, 'status', '?')) for t in tasks]) + return tasks + + def _complete_workflow(workflow_executor: WorkflowExecutor, workflow_id: str): """Complete workflow by sending required signals""" try: @@ -659,8 +692,8 @@ def scenario_signal_target_workflow(workflow_executor: WorkflowExecutor): # Start workflow workflow_id = _start_complex_workflow(workflow_executor) - # Wait and check workflow status - time.sleep(1.0) + # Wait until it is actually parked on a task, rather than a fixed sleep. + _wait_for_blocking_task(workflow_executor, workflow_id) # Debug: Check workflow status before signaling try: @@ -728,7 +761,7 @@ def scenario_signal_blocking_workflow(workflow_executor: WorkflowExecutor): logger.info('Testing signal with BLOCKING_WORKFLOW strategy...') workflow_id = _start_complex_workflow(workflow_executor) - time.sleep(0.5) + _wait_for_blocking_task(workflow_executor, workflow_id) response = workflow_executor.signal( workflow_id=workflow_id, @@ -757,7 +790,7 @@ def scenario_signal_blocking_task(workflow_executor: WorkflowExecutor): logger.info('Testing signal with BLOCKING_TASK strategy...') workflow_id = _start_complex_workflow(workflow_executor) - time.sleep(0.5) + _wait_for_blocking_task(workflow_executor, workflow_id) response = workflow_executor.signal( workflow_id=workflow_id, @@ -787,7 +820,7 @@ def scenario_signal_blocking_task_input(workflow_executor: WorkflowExecutor): logger.info('Testing signal with BLOCKING_TASK_INPUT strategy...') workflow_id = _start_complex_workflow(workflow_executor) - time.sleep(0.5) + _wait_for_blocking_task(workflow_executor, workflow_id) response = workflow_executor.signal( workflow_id=workflow_id, @@ -818,7 +851,7 @@ def scenario_signal_default_strategy(workflow_executor: WorkflowExecutor): logger.info('Testing signal with default strategy...') workflow_id = _start_complex_workflow(workflow_executor) - time.sleep(0.5) + _wait_for_blocking_task(workflow_executor, workflow_id) # Don't specify return_strategy - should default to TARGET_WORKFLOW response = workflow_executor.signal( @@ -842,7 +875,7 @@ def scenario_signal_async(workflow_executor: WorkflowExecutor): logger.info('Testing async signal...') workflow_id = _start_complex_workflow(workflow_executor) - time.sleep(0.5) + _wait_for_blocking_task(workflow_executor, workflow_id) # Send async signal (should not return response) result = workflow_executor.signal_async( @@ -863,7 +896,7 @@ def scenario_signal_to_dict_fix(workflow_executor: WorkflowExecutor): logger.info('Testing to_dict() method fix...') workflow_id = _start_complex_workflow(workflow_executor) - time.sleep(0.5) + _wait_for_blocking_task(workflow_executor, workflow_id) response = workflow_executor.signal( workflow_id=workflow_id, From 11891cb8c5be511536db2c80b0723651cd01000f Mon Sep 17 00:00:00 2001 From: Manan Bhatt Date: Thu, 13 Aug 2026 22:40:14 +0530 Subject: [PATCH 7/8] Poll to terminal instead of fixed sleeps in the last two flaky spots Task-level diagnostics from a red run showed these are two different problems, not the one I assumed: scenario_decorated_workers: task SCHEDULED, pollCount=0, workerId=None after sleep(15), while both decorated workers were demonstrably active and polling every 100ms. The server simply had not handed the task out inside that window. This is the same false negative the batch-completion budget in this file was already raised to fix, so use that budget here and poll to terminal. test_v2_fallback_intg: 4 of a workflow's 5 tasks COMPLETED with the last IN_PROGRESS on a live worker (pollCount=1, workerId set) when the 60s budget expired -- it was progressing, not stuck. Raised to 120s. A task still IN_PROGRESS after that is a real lost update rather than slowness, and the diagnostic prints it. --- tests/integration/test_v2_fallback_intg.py | 9 +++++++-- .../workflow/test_workflow_execution.py | 16 ++++++++++++++-- 2 files changed, 21 insertions(+), 4 deletions(-) diff --git a/tests/integration/test_v2_fallback_intg.py b/tests/integration/test_v2_fallback_intg.py index 442e2d0c..143f787b 100644 --- a/tests/integration/test_v2_fallback_intg.py +++ b/tests/integration/test_v2_fallback_intg.py @@ -142,8 +142,13 @@ def _run_workers(): print(f"\n Submitted {len(workflow_ids)} workflows") - # Wait for completion - deadline = time.time() + 60 # 60s timeout + # Wait for completion. 60s was marginal: a red run showed 4 of a + # workflow's 5 tasks COMPLETED and the last one still IN_PROGRESS on + # a live worker, i.e. the run was progressing when the budget ran + # out. Give the shared server the same headroom the other suites + # use. A task still IN_PROGRESS after this is a genuine stuck + # update, and the diagnostic below prints it. + deadline = time.time() + 120 pending = set(workflow_ids) completed = 0 failed = 0 diff --git a/tests/integration/workflow/test_workflow_execution.py b/tests/integration/workflow/test_workflow_execution.py index 87ebe012..b2dc8247 100644 --- a/tests/integration/workflow/test_workflow_execution.py +++ b/tests/integration/workflow/test_workflow_execution.py @@ -156,8 +156,20 @@ def scenario_decorated_workers( start_wf_req = StartWorkflowRequest(name=workflow_name, task_to_domain=td_map) workflow_id_2 = workflow_executor.start_workflow(start_wf_req) - logger.debug(f'started TestPythonDecoratedWorkerWf with domain:cool and id: {workflow_id_2}') - sleep(15) + logger.info('started TestPythonDecoratedWorkerWf %s (no domain) and %s (domain:cool)', + workflow_id, workflow_id_2) + + # Poll to terminal instead of sleeping a fixed 15s. Both of these run a + # single decorated-worker task, and on a loaded shared server the task can + # sit SCHEDULED well past 15s before the server hands it to a poller -- + # observed as status=SCHEDULED pollCount=0 while the worker was demonstrably + # alive and polling every 100ms. Same false-negative the batch-completion + # budget above was raised to fix; use that budget here too. + for wf_id in (workflow_id, workflow_id_2): + wait_for_workflow_terminal( + workflow_executor, wf_id, + timeout_seconds=WORKFLOW_COMPLETION_MAX_WAIT_SECONDS, + ) _run_with_retry_attempt( validate_workflow_status, From d84f593a335d1ee854ef1fd27f807dc5bd0b4a8e Mon Sep 17 00:00:00 2001 From: Manan Bhatt Date: Thu, 13 Aug 2026 22:59:06 +0530 Subject: [PATCH 8/8] Make the tag scenarios re-runnable retry_scenario re-runs a scenario from the top on a transient blip, but the tag sub-tests assumed a clean slate. A run showed exactly that: transient (0) in test_task_lifecycle running scenario test_task_lifecycle (attempt 2) Attempt 1 had already added tags, so attempt 2's "add one tag, now there is exactly one" assertion counted the leftovers and failed with `assert 2 == 1`. Clear the tags first so the assertions hold on any attempt. --- .../client/orkes/test_orkes_clients.py | 27 +++++++++++++++++++ 1 file changed, 27 insertions(+) diff --git a/tests/integration/client/orkes/test_orkes_clients.py b/tests/integration/client/orkes/test_orkes_clients.py index 42497261..a0107608 100644 --- a/tests/integration/client/orkes/test_orkes_clients.py +++ b/tests/integration/client/orkes/test_orkes_clients.py @@ -70,6 +70,21 @@ def _assert_not_found(fetch, *identifiers): raise AssertionError("expected a 404, but the resource is still readable") +def _clear_tags(get_tags, delete_tag, target): + # Best-effort: leave the target with no tags so a scenario that asserts on + # exact tag counts can be re-run. Failures are ignored -- a tag that is + # already gone is nothing to report, and cleanup must not fail the test. + try: + existing = get_tags(target) or [] + except Exception: + return + for tag in existing: + try: + delete_tag(MetadataTag(tag.key, tag.value), target) + except Exception: + pass + + def _await_value(fetch, expected, timeout=15, interval=1): # Queue size is eventually consistent -- the server enqueues and indexes # asynchronously, so reading straight after starting or draining work @@ -503,6 +518,14 @@ def __test_task_tags(self): MetadataTag("tag3", "val3") ] + # retry_scenario re-runs this whole scenario from the top on a transient + # blip, but tags survive the failed attempt -- so the "add one tag, now + # there is exactly one" assertion below saw the leftovers and failed with + # `assert 2 == 1`. Start from a known-empty set so the scenario is + # re-runnable. + _clear_tags(self.metadata_client.getTaskTags, + self.metadata_client.deleteTaskTag, TASK_TYPE) + self.metadata_client.addTaskTag(tags[0], TASK_TYPE) fetchedTags = self.metadata_client.getTaskTags(TASK_TYPE) assert len(fetchedTags) == 1 @@ -519,6 +542,10 @@ def __test_task_tags(self): def __test_workflow_tags(self): singleTag = MetadataTag("wftag", "val") + # Same re-runnability problem as __test_task_tags. + _clear_tags(self.metadata_client.get_workflow_tags, + self.metadata_client.delete_workflow_tag, WORKFLOW_NAME) + self.metadata_client.add_workflow_tag(singleTag, WORKFLOW_NAME) fetchedTags = self.metadata_client.get_workflow_tags(WORKFLOW_NAME) assert len(fetchedTags) == 1