From 58bab116f2452875fb52e0e200dfb5e279b62df0 Mon Sep 17 00:00:00 2001 From: Scott K Logan Date: Fri, 28 Aug 2026 15:49:15 -0500 Subject: [PATCH 1/2] Make job completion order deterministic, improve tests When multiple jobs complete in the same "tick" of the parallel execution loop, the `done_futures` set can result in different colcon return codes depending on which one happens to get processed first. A more desirable behavior is to prefer the return code of the job which was started first. While it is currently unlikely to see multiple jobs finish in the same tick due to strict concurrency pre-limiting, this becomes much more common under dynamic scheduling and resource-guarded execution. This change also includes an additional test for the parallel executor, and switches all of the job identifiers to use consistent identifiers, where they were previously using different values in the overall job dictionary from the identifier used on the job objects themselves. Assisted-by: Gemini 3.5 Flash --- colcon_parallel_executor/executor/parallel.py | 4 ++- test/test_executor_parallel.py | 27 ++++++++++++++----- 2 files changed, 23 insertions(+), 8 deletions(-) diff --git a/colcon_parallel_executor/executor/parallel.py b/colcon_parallel_executor/executor/parallel.py index f94a832..5c4cc87 100644 --- a/colcon_parallel_executor/executor/parallel.py +++ b/colcon_parallel_executor/executor/parallel.py @@ -168,7 +168,9 @@ async def _execute(self, args, jobs, *, on_error): f.identifier for f in futures.values()))) # check results of done futures - for done_future in done_futures: + for done_future in [ + f for f in futures.keys() if f in done_futures + ]: job = futures[done_future] del futures[done_future] # get result without raising an exception diff --git a/test/test_executor_parallel.py b/test/test_executor_parallel.py index 47093ae..1c6bc61 100644 --- a/test/test_executor_parallel.py +++ b/test/test_executor_parallel.py @@ -110,7 +110,7 @@ def test_parallel(): args = SimpleNamespace(parallel_workers=2) jobs = OrderedDict() - jobs['one'] = Job1() + jobs['job1'] = Job1() # success rc = extension.execute(args, jobs) @@ -119,8 +119,8 @@ def test_parallel(): ran_jobs.clear() # return error code - jobs['two'] = Job2() - jobs['four'] = Job4() + jobs['job2'] = Job2() + jobs['job4'] = Job4() rc = extension.execute(args, jobs) assert rc == 2 assert ran_jobs == ['job1'] @@ -144,22 +144,22 @@ def test_parallel(): ran_jobs.clear() # continue after error, keeping first error code - jobs['five'] = Job5() + jobs['job5'] = Job5() rc = extension.execute(args, jobs, on_error=OnError.continue_) assert rc == 2 assert ran_jobs == ['job1', 'job4'] ran_jobs.clear() # continue but skip downstream - jobs['six'] = Job6() - jobs['seven'] = Job7() + jobs['job6'] = Job6() + jobs['job7'] = Job7() rc = extension.execute(args, jobs, on_error=OnError.skip_downstream) assert rc == 2 assert ran_jobs == ['job1', 'job7', 'job4'] ran_jobs.clear() # exception - jobs['two'] = Job3() + jobs['job2'] = Job3() rc = extension.execute(args, jobs) assert isinstance(rc, RuntimeError) assert ran_jobs == ['job1'] @@ -400,3 +400,16 @@ def test_parallel_workers_zero(): assert rc == 0 assert set(ran_jobs) == {'job1', 'job2'} ran_jobs.clear() + + +def test_parallel_exception_skip_pending(): + extension = ParallelExecutorExtension() + args = SimpleNamespace(parallel_workers=1) + jobs = OrderedDict() + jobs['job3'] = Job3() + jobs['job4'] = Job4() + + rc = extension.execute(args, jobs, on_error=OnError.skip_pending) + assert isinstance(rc, RuntimeError) + assert ran_jobs == [] + ran_jobs.clear() From 7b5e008a697009f559cf73cfbc5424b2fbfaed2d Mon Sep 17 00:00:00 2001 From: Kimberly McGuire Date: Thu, 10 Sep 2026 17:35:53 -0500 Subject: [PATCH 2/2] Add test specifically to cover deterministic RC Assisted-by: Claude Opus 5.1 --- test/test_executor_parallel.py | 26 ++++++++++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/test/test_executor_parallel.py b/test/test_executor_parallel.py index 1c6bc61..4ba1d82 100644 --- a/test/test_executor_parallel.py +++ b/test/test_executor_parallel.py @@ -413,3 +413,29 @@ def test_parallel_exception_skip_pending(): assert isinstance(rc, RuntimeError) assert ran_jobs == [] ran_jobs.clear() + + +class ImmediateJob(Job): + + def __init__(self, identifier, rc): + super().__init__( + identifier=identifier, dependencies=set(), task=None, + task_context=None) + self.rc = rc + + async def __call__(self, *args, **kwargs): + return self.rc + + +def test_simultaneous_completion_prefers_first_started(): + extension = ParallelExecutorExtension() + args = SimpleNamespace(parallel_workers=2) + + # repeat: set iteration order varies run to run, so a single + # pass would only catch the bug ~50% of the time + for _ in range(50): + jobs = OrderedDict() + jobs['early'] = ImmediateJob('early', 3) + jobs['late'] = ImmediateJob('late', 7) + rc = extension.execute(args, jobs, on_error=OnError.continue_) + assert rc == 3, f'expected first-started rc 3, got {rc}'