diff --git a/tests/integration/client/orkes/test_orkes_clients.py b/tests/integration/client/orkes/test_orkes_clients.py index 2811d492..a0107608 100644 --- a/tests/integration/client/orkes/test_orkes_clients.py +++ b/tests/integration/client/orkes/test_orkes_clients.py @@ -53,6 +53,51 @@ 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 _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 + # 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 +174,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 +205,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 +249,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 +297,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 +373,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 +507,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 = [ @@ -496,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 @@ -512,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 @@ -575,11 +609,9 @@ def __test_workflow_execution_lifecycle(self): assert workflow.status == "RUNNING" self.workflow_client.delete_workflow(workflow_uuid) - try: - 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) + _assert_not_found( + lambda: self.workflow_client.get_workflow(workflow_uuid, False), workflow_uuid + ) def __test_task_execution_lifecycle(self): @@ -606,7 +638,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 @@ -617,7 +651,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, @@ -666,7 +702,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 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..4270e190 --- /dev/null +++ b/tests/integration/leaked_task_defs.py @@ -0,0 +1,111 @@ +"""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. +""" + +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) diff --git a/tests/integration/test_v2_fallback_intg.py b/tests/integration/test_v2_fallback_intg.py index b265da79..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 @@ -171,6 +176,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..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, @@ -239,6 +251,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 +285,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, @@ -603,6 +644,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: @@ -630,8 +704,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: @@ -699,7 +773,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, @@ -728,7 +802,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, @@ -758,7 +832,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, @@ -789,7 +863,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( @@ -813,7 +887,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( @@ -834,7 +908,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,