#6809 WIP: Fix stuck workflow nodes that are starved by map tasks exceeding the max parallelism budget Pull request opened by fg91 on <!date^1766055240^{date_short}|2025-12-18T10:54:00Z> ## Why are the changes needed? In this minimal reproducible example ... from flytekit import task, workflow, map_task import time @task def sleep(inp: int) -> None: time.sleep(inp) @workflow def wf() -> None: map_task(sleep)(inp=[3600 for _ in range(30)]) # array node with long running tasks. 30 is chosen to exceed default max parallelism of 25 sleep(inp=30) # Task that runs more quickly ... I observe the following and in my opinion undesirable behaviour: • Both the map task and the node after it start running. • The

n1

pod succeeds quickly: [Screenshot 2025-12-18 at 11 44 04](https://private-user-images.githubusercontent.com/36511035/528052985-03377611-4f97-40ec-8acc-1d5632e8abf3.png?jwt=eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3NzE1OTg0MDAsIm5iZiI6MTc3MTU5ODEwMCwicGF0aCI6Ii8zNjUxMTAzNS81MjgwNTI5ODUtMDMzNzc2MTEtNGY5Ny00MGVjLThhY2MtMWQ1NjMyZThhYmYzLnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNjAyMjAlMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjYwMjIwVDE0MzUwMFomWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPTIyZDY0YjFiYzM4OTg3YzQ2ZjVkNGU3NThiZjhjMGVjMjNhYWY5NDk1ZWQ5Y2I1NGJlNGI1NjFiOTZmMDYyYWEmWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0In0.NRDVht2gRSl3ULBIL2RP57fscDmTZXETInJHK926dn8) • The

n1

node stays in

Queued

state in the Flyte UI despite the underlying pod already having succeeded: [Screenshot 2025-12-18 at 11 43 45](https://private-user-images.githubusercontent.com/36511035/528052709-61b00cb6-4ef9-4dc7-a1f7-e55127c59b3a.png?jwt=eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJpc3MiOiJnaXRodWIuY29tIiwiYXVkIjoicmF3LmdpdGh1YnVzZXJjb250ZW50LmNvbSIsImtleSI6ImtleTUiLCJleHAiOjE3NzE1OTg0MDAsIm5iZiI6MTc3MTU5ODEwMCwicGF0aCI6Ii8zNjUxMTAzNS81MjgwNTI3MDktNjFiMDBjYjYtNGVmOS00ZGM3LWExZjctZTU1MTI3YzU5YjNhLnBuZz9YLUFtei1BbGdvcml0aG09QVdTNC1ITUFDLVNIQTI1NiZYLUFtei1DcmVkZW50aWFsPUFLSUFWQ09EWUxTQTUzUFFLNFpBJTJGMjAyNjAyMjAlMkZ1cy1lYXN0LTElMkZzMyUyRmF3czRfcmVxdWVzdCZYLUFtei1EYXRlPTIwMjYwMjIwVDE0MzUwMFomWC1BbXotRXhwaXJlcz0zMDAmWC1BbXotU2lnbmF0dXJlPTY0ZGRiNjk5NzJkZmRiZWZkZDU2ODNkN2NiNTg5OWExYTliMDkwZjgyZTM2MjllOGI3ODU5MGQ1YmYzNzFjMWEmWC1BbXotU2lnbmVkSGVhZGVycz1ob3N0In0.lYpCdSniw-bSFqH-tpKOxOR-PBivU8dlvR3DzIXqFNw) • The

n1

node is updated in the Flyte UI from

Queued

to

Succeeded

only after the map task (or at least enough tasks within it) completes as well which could be hours later. ### Reason for this behaviour • In the `RecursiveNodeHandler` which traverses the workflow graph, we check whether the current degree of parallelism has exceeded the max parallelism: func (c *recursiveNodeExecutor) RecursiveNodeHandler(ctx context.Context, execContext executors.ExecutionContext, dag executors.DAGStructure, nl executors.NodeLookup, currentNode v1alpha1.ExecutableNode) ( interfaces.NodeStatus, error) { ... if IsMaxParallelismAchieved(ctx, currentNode, nodePhase, execContext) { ... return interfaces.NodeStatusRunning, nil • For

n1

,

IsMaxParallelismAchieved

gives `true`: • In the example above, when evaluating

n1

, the current degree of parallelism takes a value of

31

which is bigger than the default max parallelism of

25

. • The degree of parallelism is increased by 1 for the array node itself here and 30 times for each of the tasks in the map task here. • This means that the task phase of

n1

will be evaluated only once less than 25 tasks are still running within the map task

n0

. ## What changes were proposed in this pull request? • As a user, I would expect

n1

to be marked as succeeded immediately after the pod completes and not hours later when enough of the array node tasks complete. • Alternatively, if there is no "parallelism budget" for

n1

, I would expect

n1

to not start at all until

n0

is done. But as a user I wouldn't expect this "mixed" behaviour with a node that is seemingly stuck for hours despite having completed. ## Discussion The behaviour can be avoided when modifying the parallelism tracking logic to count the map task as

1

and not as

1 + 30

(in this example). I would like to discuss which of the two is the intended behaviour. ## How was this patch tested? ## Check all the applicable boxes • I updated the documentation accordingly. • All new and existing tests passed. • All commits are signed-off. ## Related PRs ## Docs link flyteorg/flyte