Canvas starting with a group does not respect a countdown applied to whole canvas

Checklist

• I have verified that the issue exists against the master branch of Celery.
• This has already been asked to the discussions forum first.
• I have read the relevant section in the contribution guide on reporting bugs.
• I have checked the issues list for similar or identical bug reports. (#4405 may be related, but I'm not sure)
• I have checked the pull requests list for existing proposed fixes.
• I have checked the commit log to find out if the bug was already fixed in the master branch.
• I have included all related issues and possible duplicate issues in this issue (If there are none, check this box anyway). #4405 may be related, as mentioned above

Mandatory Debugging Information

• I have included the output of celery -A proj report in the issue. (if you are not able to do this, then at least specify the Celery version affected).
• I have verified that the issue exists against the master branch of Celery.
• I have included the contents of pip freeze in the issue.
• I have included all the versions of all the external dependencies required to reproduce this bug.

Optional Debugging Information

• I have tried reproducing the issue on more than one Python version and/or implementation.
• I have tried reproducing the issue on more than one message broker and/or result backend.
• I have tried reproducing the issue on more than one version of the message broker and/or result backend.
• I have tried reproducing the issue on more than one operating system.
• I have tried reproducing the issue on more than one workers pool.
• I have tried reproducing the issue with autoscaling, retries, ETA/Countdown & rate limits disabled.
• I have tried reproducing the issue after downgrading and/or upgrading Celery and its dependencies.

Related Issues and Possible Duplicates

Related Issues

• #4405

Possible Duplicates

• None

Environment & Settings

Celery version: 5.3.0b1 (dawn-chorus)

software -> celery:5.3.0b1 (dawn-chorus) kombu:5.3.0b2 py:3.8.12 billiard:4.0.2 redis:4.1.4 platform -> system:Linux arch:64bit, ELF kernel version:5.4.0-131-generic imp:CPython loader -> celery.loaders.app.AppLoader settings -> transport:redis results:redis:///

accept_content: ['json', 'pickle'] event_serializer: 'json' result_serializer: 'pickle' task_serializer: 'pickle' broker_url: 'redis://localhost:6379//' deprecated_settings: None result_backend: 'redis:///'

Steps to Reproduce

Required Dependencies

• Minimal Python Version: N/A or Unknown
• Minimal Celery Version: N/A or Unknown
• Minimal Kombu Version: N/A or Unknown
• Minimal Broker Version: N/A or Unknown
• Minimal Result Backend Version: N/A or Unknown
• Minimal OS and/or Kernel Version: N/A or Unknown
• Minimal Broker Client Version: N/A or Unknown
• Minimal Result Backend Client Version: N/A or Unknown

Python Packages

adc-streaming==2.0.0 amqp==5.1.1 astroid==2.4.2 backports.entry-points-selectable==1.1.0 billiard==4.0.2 cachy==0.3.0 celery @ git+https://github.com/celery/celery.git@e0b0af6c7af9f7a127ae0321dc4e798433c89592 certifi==2021.10.8 cffi==1.15.1 charset-normalizer==2.1.1 click==8.1.3 click-didyoumean==0.3.0 click-plugins==1.1.1 click-repl==0.2.0 confluent-kafka==1.8.2 crashtest==0.3.1 cryptography==36.0.2 Deprecated==1.2.13 distlib==0.3.2 filelock==3.0.12 future==0.18.2 hop-client==0.5.0 idna==3.4 igwn-auth-utils==0.3.1 isort==5.6.3 kombu==5.3.0b2 lazy-object-proxy==1.4.3 ligo-gracedb==2.8.0 lockfile==0.12.2 mccabe==0.6.1 msgpack==1.0.2 packaging==21.3 pastel==0.2.1 pkginfo==1.7.1 platformdirs==2.3.0 pluggy==1.0.0 prompt-toolkit==3.0.28 pycparser==2.21 PyJWT==2.6.0 pylev==1.4.0 pylint==2.6.0 pyparsing==3.0.7 pytz==2021.3 redis==4.1.4 requests==2.28.1 safe-netrc==1.0.0 scitokens==1.7.2 shellingham==1.4.0 six==1.16.0 toml==0.10.2 tqdm==4.62.3 typing_extensions==4.4.0 urllib3==1.26.12 vine==5.0.0 virtualenv==20.7.2 wcwidth==0.2.5 wrapt==1.13.3 xmltodict==0.12.0

Other Dependencies

N/A

Minimally Reproducible Test Case

Project tree

$ tree proj/
proj/
├── __init__.py
└── tasks.py

File contents

$ cat proj/__init__.py 
from celery import Celery

app = Celery('tasks', broker = 'redis://')
app.conf['result_backend'] = app.conf.broker_url
app.add_defaults(
    dict(
        accept_content = ['json', 'pickle'],
        event_serializer = 'json',
        result_serializer = 'pickle',
        task_serializer = 'pickle'
    )
)
app.autodiscover_tasks([__name__])

$ cat proj/tasks.py 
from . import app


@app.task
def add(x, y):
    return x + y


@app.task
def prod(x, y):
        return x * y

Start worker with

$ celery -A proj worker -l info -c 2

Then run following script

from time import time

from celery import group

from proj import tasks

canvas = group(
    tasks.add.si(1, 2),
    tasks.add.si(3, 4)
)

canvas |= tasks.add.si(5, 6)
canvas |= tasks.add.si(7, 8)
canvas |= tasks.add.si(9, 10)


t = time()
r = canvas.apply_async(countdown=10)
print("Result = ", r.get())
print("Time elapsed(s)", time() -t)

Expected Behavior

Since a countdown has been placed, no task should execute until the countdown has expired.

Actual Behavior

If a canvas starts with a group, and has chained tasks, the group is executed right away. For example in my minimal script above, there is the following canvas

canvas = group(
    tasks.add.si(1, 2),
    tasks.add.si(3, 4)
)

canvas |= tasks.add.si(5, 6)
canvas |= tasks.add.si(7, 8)
canvas |= tasks.add.si(9, 10)


t = time()
r = canvas.apply_async(countdown=10)

The group containing the two add tasks are evaluated rightaway in the queue. The timeout kicks in only after this. This is seen in the log of the worker (see timestamp of result 3, 7 vs. 11, 15, 19):

[2022-10-24 23:10:04,176: INFO/MainProcess] mingle: searching for neighbors
[2022-10-24 23:10:04,177: WARNING/MainProcess] No hostname was supplied. Reverting to default 'localhost'
[2022-10-24 23:10:05,183: INFO/MainProcess] mingle: all alone
[2022-10-24 23:10:05,201: INFO/MainProcess] celery@deep-ThinkPad-P53 ready.
[2022-10-24 23:13:33,585: INFO/MainProcess] Task proj.tasks.add[ce4dfe2e-6188-4762-ae17-8bcda6c6124a] received
[2022-10-24 23:13:33,586: INFO/MainProcess] Task proj.tasks.add[e89c93a0-ae5f-482e-a5a2-6afb4e8168d3] received
[2022-10-24 23:13:33,589: INFO/ForkPoolWorker-1] Task proj.tasks.add[ce4dfe2e-6188-4762-ae17-8bcda6c6124a] succeeded in 0.002615050001622876s: 3
[2022-10-24 23:13:33,591: WARNING/ForkPoolWorker-2] No hostname was supplied. Reverting to default 'localhost'
[2022-10-24 23:13:33,594: INFO/ForkPoolWorker-2] Task proj.tasks.add[e89c93a0-ae5f-482e-a5a2-6afb4e8168d3] succeeded in 0.00721587799853296s: 7
[2022-10-24 23:13:33,594: INFO/MainProcess] Task proj.tasks.add[864c3c54-5569-4a82-98d7-7aceaf9f4eee] received
[2022-10-24 23:13:43,597: WARNING/ForkPoolWorker-1] No hostname was supplied. Reverting to default 'localhost'
[2022-10-24 23:13:43,601: INFO/ForkPoolWorker-1] Task proj.tasks.add[864c3c54-5569-4a82-98d7-7aceaf9f4eee] succeeded in 0.008149747001880314s: 11
[2022-10-24 23:13:43,601: INFO/MainProcess] Task proj.tasks.add[a6d1d655-5cae-4248-bdeb-5844e8c3efce] received
[2022-10-24 23:13:43,602: INFO/ForkPoolWorker-1] Task proj.tasks.add[a6d1d655-5cae-4248-bdeb-5844e8c3efce] succeeded in 0.0008764769991103094s: 15
[2022-10-24 23:13:43,603: INFO/MainProcess] Task proj.tasks.add[ff2fa778-d6ef-4976-ac21-f46ebd9dd88d] received
[2022-10-24 23:13:43,603: INFO/ForkPoolWorker-1] Task proj.tasks.add[ff2fa778-d6ef-4976-ac21-f46ebd9dd88d] succeeded in 0.00035578700044425204s: 19

The tasks starting ce4dfe2e and e89c93a0 are received and are evaluated immediately giving results 3 and 7. The tasks that are chained to the group evaluate after the timeout is expired (note the timing for the task starting 864c3c54).

A few other observations

• If there are chained tasks after the group, then setting countdown to the group directly is not respected. For example, changing minimal example to

from time import time

from celery import group

from proj import tasks

canvas = group(
    tasks.add.si(1, 2),
    tasks.add.si(3, 4)
).set(countdown=10)  # set countdown here

canvas |= tasks.add.si(5, 6)
canvas |= tasks.add.si(7, 8)
canvas |= tasks.add.si(9, 10)


t = time()
r = canvas.apply_async()
print("Result = ", r.get())
print("Time elapsed(s)", time() -t)

Gives all results rightaway:

Result =  19
Time elapsed(s) 0.03932476043701172

However, if there are no chained tasks i.e., just the group, then the behavior is as expect. For example,

from time import time

from celery import group

from proj import tasks

canvas = group(
    tasks.add.si(1, 2),
    tasks.add.si(3, 4)
).set(countdown=10)

t = time()
r = canvas.apply_async()
print("Result = ", r.get())
print("Time elapsed(s)", time() -t)

will print

Result =  [3, 7]
Time elapsed(s) 10.034985542297363