Possible issue with pipelines starting with a group

[jberclaz]

Issue

I've been experiencing an issue with Celery pipelines involving group and I believe it may be the symptom of a bug.

I'm creating a pipeline starting with a group, followed by a few serial tasks:

pipeline = chord([task1.s() for i in range(3)], chain(task2.s(), task3.s()))

The same pipeline can also be declared as:

pipeline = chain(group([task1.s() for i in range(3)]), task2.s(), task3.s())

In both cases, the pipeline executes properly. However, when calling pipeline.apply_async(), the returned AsyncResult does not contain the whole pipeline and skips task2. In other words, apply_async returns an AsyncResult representing task3, whose parent is the group task.

Workaround

I found a workaround to this issue, by adding a single task in front of the pipeline:

pipeline = chain(task0.s(), group([task1.s() for i in range(3)]), task2.s(), task3.s())

In this case, calling pipeline.apply_async() returns the correct chain of tasks, i.e. task3 -> task2 -> group_task -> task0.

Version

I'm using Celery 5.2.7 on Linux with RabbitMQ as broker and Redis as backend.

[Nusnus]

Hey @jberclaz, I think this might be related to the fix from #5958 which will be part of the 5.3 release.
So to confirm your case, can you try to reproduce the issue with the latest main branch please?

Thank you!

[jberclaz]

Hi @Nusnus , thanks for your answer.

I tried with the latest from main, but as far as I can see, the problem still occurs.

Here is what I did to test it:

1. I cloned the latest main branch
2. I built the wheel package python3 setup.py bdist_wheel
3. I installed the package in my virtualenv and ran my test.

[Nusnus]

I have a lot on my plate right now so I can't promise I'll get to this soon, but it's good to confirm the bug exists, thank you.

You're welcome to attempt to fix it yourself if you wish and send me a PR for review.
If its a bit too much, I'd recommend try implementing an integration test that reproduces this issue under the testing environment.

Posting here such a test case can accelerate the fix, but the given workaround can also be enough for the moment.

P.S
Sorry for the late replay, as I said I have a bit of a busy time right now :)

[jberclaz]

Thanks for following up, and no worries about the delay.

The integration test is a good suggestion. I'll try to write one as soon as I can.

[Nusnus]

Awesome!
Thanks ❤️

Let me know if you need a hand ;)

[jberclaz]

Hi @Nusnus,

I wrote an integration test that highlights the issue. What's the best way to communicate it? Append it to this thread? Push a branch?

Thanks

[Nusnus]

Great @jberclaz !
To review it, well, yeah you can post it here, but opening a pull request would make it easier to discuss and develop where we can leave comments on code and make sure nothing else is broken (via the CI).

TL;DR - you need to have a celery fork, create your branch there, commit the changes and then open a PR to main from your fork/branch and request me as a reviewer.

[jberclaz]

Sounds good. Here is the PR.

[Nusnus]

Fixed in main: #7919, #8663