A pool child lost after a task's link is published, before its result is stored, fires the error callback beside the link #10952

aviadr1 · 2026-10-25T07:55:49Z

aviadr1
Oct 25, 2026

Note

How this was found, and an offer. This issue came from pytest-obligation, an open-source pytest plugin (Apache-2.0) for background work that must not be lost, run twice or recorded wrongly. You describe one task and how to see what it did. The plugin then generates the failure cases: on a real celery worker it kills the pool child at each of Celery's stages, makes a hook raise, refuses a publish, restarts the worker, and compares every outcome with a normal run.

In Celery it has produced #10724 (fixed by #10752 and #10759), #10725 and #10949. Each finding stays in the suite as a pinned outcome and a strict xfail, so a fix shows up as a change, and so does a regression. That already happened once: after main merged the two #10724 fixes, the unchanged contract flagged exactly those two cases as changed. The harness also grows. Since the first version of #10726 it has added a chord history and two more guarantee families. Each new release runs the same contract against more ways to fail.

We'd like Celery to adopt it. #10726 does that fully: the contract in t/due_work, plus a CI job (Linux, one Redis service, about 3.5 minutes with install, and a test-only dependency). If you want it, a follow-up PR can add pytest-obligation's PR report, a single comment on each PR summarizing what the contract proved.

Disclosure: the harness is written by Aviad Rozenhek (CTO, tapin.io), who directed this work. This report, the reproduction and the PR were written with Claude Code, and every result in them was produced by running the code.

Checklist

I have verified that the issue exists against the main 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.
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 main branch.
I have included all related issues and possible duplicate issues
in this issue (If there are none, check this box anyway).
I have tried to reproduce the issue with pytest-celery and added the reproduction script below.

Mandatory Debugging Information

I have included the output of celery -A proj report in the issue.
I have verified that the issue exists against the main branch of Celery.
I have included the contents of pip freeze in the issue (the packages that matter; the environment also has requirements/test.txt).
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

A task whose SUCCESS is stored still fires its error callback: a pool child lost after the store, or a raising on_success #10724: a pool child lost after SUCCESS was stored fired the error callback beside the link. Don't fail a task whose SUCCESS was stored before the worker was lost #10752 fixed that by checking the stored state. This issue is the same symptom one step earlier: the link is already out, and nothing has been stored yet, so that check has nothing to find.
A link the broker refuses after the task body ran records FAILURE and acks the message, so the link is never sent #10725: a refused link records FAILURE for a task whose body ran.
Error in collect with group #1936: why links are published before the result is stored (so result.children is populated).

Possible Duplicates

None

Environment & Settings

Celery version: 5.7.0rc1, main at e3cc7cc

celery report Output:

software -> celery:5.7.0rc1 kombu:5.7.0rc1 py:3.13.13
            billiard:4.3.1 redis:8.1.0
platform -> system:Linux arch:64bit, ELF
            kernel version:6.18.54.1-microsoft-standard-WSL2 imp:CPython
loader   -> celery.loaders.app.AppLoader
settings -> transport:redis results:redis://localhost:6379/11

broker_url: 'redis://localhost:6379/10'
result_backend: 'redis://localhost:6379/11'
task_acks_late: True
worker_prefetch_multiplier: 1

Steps to Reproduce

Required Dependencies

Minimal Python Version: N/A or Unknown (reproduced on 3.13.13)
Minimal Celery Version: N/A or Unknown (reproduced on main, e3cc7cc)
Minimal Kombu Version: N/A or Unknown (5.7.0rc1)
Minimal Broker Version: N/A or Unknown (Redis 7.4.11)
Minimal Result Backend Version: N/A or Unknown (Redis 7.4.11)
Minimal OS and/or Kernel Version: N/A or Unknown (Linux)
Minimal Broker Client Version: N/A or Unknown (redis-py 8.1.0)
Minimal Result Backend Client Version: N/A or Unknown (redis-py 8.1.0)

Python Packages

pip freeze Output:

amqp==5.4.1
billiard==4.3.1
celery @ file:///src (main, e3cc7cc53)
click==8.5.0
kombu==5.7.0rc1
redis==8.1.0
vine==5.1.0

Other Dependencies

Details

Redis 7.4.11 as broker and result backend.

Minimally Reproducible Test Case

One task, send_message, with a link (announce) and an error callback (report_failure), task_acks_late=True, celery worker --pool=prefork --concurrency=1. The pool child exits with os._exit(1) when the tracer calls mark_as_done, which is after the link is published. Then the worker is started again. Needs Redis on localhost:6379; run python repro.py, or REJECT=1 python repro.py for task_reject_on_worker_lost=True.

repro.py

"""
A pool child lost after the task's link is published, before its result is stored.

Needs Redis on localhost:6379 (override with REDIS). Run: python repro.py
With REJECT=1 the worker runs with task_reject_on_worker_lost=True.
"""

import os
import subprocess
import sys
import time

import redis

from celery import Celery
from celery.result import AsyncResult

REDIS = os.environ.get('REDIS', 'redis://localhost:6379')
RECORDS = redis.Redis.from_url(f'{REDIS}/12', decode_responses=True)

app = Celery('repro', broker=f'{REDIS}/10', backend=f'{REDIS}/11')
app.conf.update(
    task_acks_late=True,
    worker_prefetch_multiplier=1,
    task_reject_on_worker_lost=os.environ.get('REJECT') == '1',
)


@app.task
def send_message(message):
    RECORDS.rpush('sent', message)
    return message


@app.task
def announce(result):
    """The link."""
    RECORDS.rpush('announced', f'sent {result!r}')


@app.task
def report_failure(request, exc, traceback):
    """The error callback."""
    RECORDS.rpush('announced', f'failed: {type(exc).__name__}')


def die_before_the_store():
    """In the worker: the first time send_message's result is stored, the pool child exits instead."""
    backend = type(app.backend)
    store = backend.mark_as_done

    def mark_as_done(self, task_id, result, request=None, *args, **kwargs):
        if getattr(request, 'task', None) == send_message.name and RECORDS.set('died', 1, nx=True):
            os._exit(1)  # the link was already published, just above mark_as_done in trace.py
        return store(self, task_id, result, request, *args, **kwargs)

    backend.mark_as_done = mark_as_done


def worker():
    return subprocess.Popen([sys.executable, __file__, 'worker'])


def wait(task_id):
    deadline = time.monotonic() + 30
    while not AsyncResult(task_id, app=app).ready() and time.monotonic() < deadline:
        time.sleep(0.2)
    time.sleep(2)  # let the link and the error callback run


def stop(process):
    process.terminate()
    process.wait(30)


if __name__ == '__main__' and sys.argv[1:] == ['worker']:
    die_before_the_store()
    app.worker_main(['worker', '--pool=prefork', '--concurrency=1', '--loglevel=WARNING',
                     '--without-mingle', '--without-gossip', '--without-heartbeat'])
elif __name__ == '__main__':
    for db in (10, 11):
        redis.Redis.from_url(f'{REDIS}/{db}').flushdb()
    RECORDS.flushdb()
    first = worker()
    task_id = send_message.apply_async(
        ('your order has shipped',), link=announce.s(), link_error=report_failure.s()
    ).id
    wait(task_id)
    stop(first)
    second = worker()  # recovery: the worker starts again
    time.sleep(5)
    wait(task_id)
    stop(second)
    print('state:    ', AsyncResult(task_id, app=app).state)
    print('sent:     ', RECORDS.lrange('sent', 0, -1))
    print('announced:', sorted(RECORDS.lrange('announced', 0, -1)))

$ python repro.py
state:     FAILURE
sent:      ['your order has shipped']
announced: ['failed: WorkerLostError', "sent 'your order has shipped'"]

$ REJECT=1 python repro.py
state:     SUCCESS
sent:      ['your order has shipped', 'your order has shipped']
announced: ["sent 'your order has shipped'", "sent 'your order has shipped'"]

Expected Behavior

One execution reports one outcome. Either the link runs or the error callback runs, not both. The recorded state agrees with whichever callback ran. In a chord, a header whose body ran is not counted as a failure while its link went out as a success.

Actual Behavior

trace.py publishes the task's links and chain, then stores the result. The order is deliberate (#1936). If the pool child is lost between the two, there is nothing stored. On an OOM kill, a SIGKILL, or a crash in mark_as_done itself (serialising a large result, for example), the parent gets WorkerLostError. Request.on_failure finds no stored SUCCESS, so #10752's check does not apply, and it records FAILURE.

Default settings: the task is recorded FAILURE and its error callback fires, but its link already announced success. The application is told both, for one execution.
task_reject_on_worker_lost=True: the message is requeued and the task runs again. That much is the documented at-least-once trade. The rerun publishes the link again, so the link runs twice: once for the lost execution and once for the rerun.
In a chord (Redis backend, two send_message headers, the first header's child lost at the same point): both messages are sent, the header is recorded FAILURE, and the chord fails with ChordError. The body never runs.

What it costs:

An application gets "sent" and "failed" for the same order.
With task_reject_on_worker_lost, the follow-up step runs twice, though only one execution's result is recorded.
A chord fails though every header's work was done.

This came out of pytest-obligation (formerly due-work-harness), the harness in #10726. Its Celery worker integration fails one task at each of Celery's stages on a real worker, restarts the worker, and compares each outcome with a normal run. On today's main the run showed #10752 and #10759 working, and this stage still diverging, for a single task and for a chord. #10726 pins it in FINDINGS and CHORD_FINDINGS as the died at mark_as_done row.

aviadr1 · 2026-10-25T09:03:19Z

aviadr1
Oct 25, 2026
Author

Reproduced with a real `SIGKILL` from outside, not only `os._exit`

The reproduction in the issue makes the pool child call os._exit(1) inside mark_as_done. To rule out an artefact of that, I paused the child at the same point and killed it from another process with kill -9, as the OOM killer would. Billiard takes its signal path here: the parent logged Process 'ForkPoolWorker-1' ... exited with 'signal 9 (SIGKILL)' and WorkerLostError('Worker exited prematurely: signal 9 (SIGKILL) Job: 0.'), where os._exit(1) gives exitcode 1. The outcome is the same:

Pool child killed at	Setting	State	Messages sent	Callbacks
`mark_as_done` (link already published)	default	FAILURE	1	`failed: WorkerLostError` and `sent`
`mark_as_done`	`task_reject_on_worker_lost=True`	SUCCESS	2	`sent` twice
`task_postrun` (SUCCESS stored)	default	SUCCESS	1	`sent`: #10752 holds
`task_postrun`	`task_reject_on_worker_lost=True`	SUCCESS	1	`sent`: #10752 holds

Celery main (e3cc7cc), Python 3.13, Redis 7.4.11. Run: python probe_sigkill.py mark_as_done (REJECT=1 for the second row, task_postrun for the last two).

probe_sigkill.py

"""
Probe 1: a real external SIGKILL of the pool child, at mark_as_done (#10949) or task_postrun (#10724).

The pool child pauses at the stage (first time only) and writes its pid; the driver kills it with
SIGKILL from outside, as the OOM killer would. Usage: python probe_sigkill.py mark_as_done|task_postrun
REJECT=1 sets task_reject_on_worker_lost=True. REDIS defaults to redis://localhost:6379.
"""

import os
import signal
import subprocess
import sys
import time

import redis

from celery import Celery, signals
from celery.result import AsyncResult

REDIS = os.environ.get('REDIS', 'redis://localhost:6379')
RECORDS = redis.Redis.from_url(f'{REDIS}/12', decode_responses=True)
PIDFILE = '/tmp/celery-probe-child.pid'

app = Celery('probe', broker=f'{REDIS}/10', backend=f'{REDIS}/11')
app.conf.update(
    task_acks_late=True,
    worker_prefetch_multiplier=1,
    task_reject_on_worker_lost=os.environ.get('REJECT') == '1',
)


@app.task
def send_message(message):
    RECORDS.rpush('sent', message)
    return message


@app.task
def announce(result):
    RECORDS.rpush('announced', f'sent {result!r}')


@app.task
def report_failure(request, exc, traceback):
    RECORDS.rpush('announced', f'failed: {type(exc).__name__}: {exc}')


def pause_here():
    if RECORDS.set('paused', 1, nx=True):
        with open(PIDFILE, 'w') as f:
            f.write(str(os.getpid()))
        time.sleep(30)  # the driver kills us long before this ends


def install(stage):
    if stage == 'mark_as_done':
        backend = type(app.backend)
        store = backend.mark_as_done

        def mark_as_done(self, task_id, result, request=None, *args, **kwargs):
            if getattr(request, 'task', None) == send_message.name:
                pause_here()  # the link was already published, just above mark_as_done in trace.py
            return store(self, task_id, result, request, *args, **kwargs)

        backend.mark_as_done = mark_as_done
    elif stage == 'task_postrun':

        def postrun(sender=None, **_):
            if getattr(sender, 'name', None) == send_message.name:
                pause_here()  # SUCCESS is stored

        signals.task_postrun.connect(postrun, weak=False)
    else:
        raise SystemExit(f'unknown stage {stage}')


def worker(stage=None):
    args = [sys.executable, __file__, 'worker'] + ([stage] if stage else [])
    return subprocess.Popen(args)


def wait(task_id, timeout=40):
    deadline = time.monotonic() + timeout
    while not AsyncResult(task_id, app=app).ready() and time.monotonic() < deadline:
        time.sleep(0.2)
    time.sleep(2)


def stop(process):
    process.terminate()
    process.wait(30)


if __name__ == '__main__' and sys.argv[1] == 'worker':
    if len(sys.argv) > 2:
        install(sys.argv[2])
    app.worker_main(['worker', '--pool=prefork', '--concurrency=1', '--loglevel=WARNING',
                     '--without-mingle', '--without-gossip', '--without-heartbeat'])
elif __name__ == '__main__':
    stage = sys.argv[1]
    for db in (10, 11):
        redis.Redis.from_url(f'{REDIS}/{db}').flushdb()
    RECORDS.flushdb()
    if os.path.exists(PIDFILE):
        os.remove(PIDFILE)
    first = worker(stage)
    task_id = send_message.apply_async(
        ('your order has shipped',), link=announce.s(), link_error=report_failure.s()
    ).id
    deadline = time.monotonic() + 40
    while not os.path.exists(PIDFILE) and time.monotonic() < deadline:
        time.sleep(0.1)
    time.sleep(0.3)
    pid = int(open(PIDFILE).read())
    print(f'state before kill: {AsyncResult(task_id, app=app).state}; killing pool child {pid} with SIGKILL')
    os.kill(pid, signal.SIGKILL)
    wait(task_id)
    stop(first)
    second = worker()  # recovery: the worker starts again
    time.sleep(5)
    wait(task_id)
    stop(second)
    meta = AsyncResult(task_id, app=app)
    print('state:    ', meta.state, '| result:', repr(meta.result)[:120])
    print('sent:     ', RECORDS.lrange('sent', 0, -1))
    print('announced:', sorted(RECORDS.lrange('announced', 0, -1)))

Written with Claude Code, directed by Aviad Rozenhek (author of pytest-obligation); every result above came from running the scripts.

0 replies

auvipy · 2026-10-25T09:27:14Z

auvipy
Oct 25, 2026
Maintainer

This is a fully AI generated code based project with not enough star and external user. so this need to be mature and community vetted first.

1 reply

aviadr1 Oct 25, 2026
Author

hi @auvipy this is Aviad, the maintainer of pytest-obligation, much appreciate of the review.
yes, you are right the project is new and star count is low,
but it is incredibly powerful, and it already found a bunch of defects in celery, rq, wagtail and many other OSS projects, not to mention we dogfood it internally both for gigaverse-app BE applications and our homegrown tasks framework gigaverse-app/due

the problem with adoption is always the same - since its heavily AI generated and new, people distrust it - at the cost of losing valuable defects it can detect.

I am looking for a first adopter and I was hoping that since it already pinpointed a few celery defects. you guys might be the ones that break the cycle.

I'd love to continue this disucssion and see what it would take to vett it.
Best,
Aviad Rozenhek, CTO tapin.io
aviad at gigaverse dot com

aviadr1 · 2026-10-25T10:43:44Z

aviadr1
Oct 25, 2026
Author

The bugs stand without the tool. Each report below now has a reproduction that needs only Celery and Redis:

Report	What happens	Status
#10724	a pool child lost after SUCCESS is stored, or a raising `on_success`, fires the error callback	fixed by @gyanu2507 in #10752 and #10759
#10725	a refused link (e.g. a full Redis broker) records FAILURE, acks the message, and the link is never sent; the same for a chord's body (comment)	open
#10954	a pool child lost after the link is published, before the result is stored, fires the error callback beside the link; with `task_reject_on_worker_lost` the link runs twice	open (the bug from this discussion, filed again as a plain report)

Each was also checked against the real failure, not only an injected one: an outside kill -9 of the pool child, and a Redis broker that really refuses writes. The outcomes were identical.

To demo the pytest-obligation integration, Its contract now runs on gigaverse-app/celery against this fork and, nightly, against Celery's current main (first run: 10 passed, 2 known failures, which are #10725 and #10954). When it finds something new, I'll file let you guys know.

0 replies

Navigation Menu

Repository navigation

A pool child lost after a task's link is published, before its result is stored, fires the error callback beside the link #10952

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{editor}}'s edit

{{editor}}'s edit

Uh oh!

Replies: 3 comments 1 reply

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{title}}

Uh oh!

Uh oh!

{{editor}}'s edit

{{editor}}'s edit

Uh oh!

Select a reply

Uh oh!

A pool child lost after a task's link is published, before its result is stored, fires the error callback beside the link #10952

Uh oh!

Uh oh!

aviadr1 Oct 25, 2026

Checklist

Mandatory Debugging Information

Optional Debugging Information

Related Issues and Possible Duplicates

Related Issues

Possible Duplicates

Environment & Settings

Steps to Reproduce

Required Dependencies

Python Packages

Other Dependencies

Minimally Reproducible Test Case

Expected Behavior

Actual Behavior

Replies: 3 comments · 1 reply

Uh oh!

aviadr1 Oct 25, 2026 Author

Reproduced with a real SIGKILL from outside, not only os._exit

Uh oh!

auvipy Oct 25, 2026 Maintainer

Uh oh!

aviadr1 Oct 25, 2026 Author

Uh oh!

Uh oh!

aviadr1 Oct 25, 2026 Author

aviadr1
Oct 25, 2026

Replies: 3 comments 1 reply

aviadr1
Oct 25, 2026
Author

Reproduced with a real `SIGKILL` from outside, not only `os._exit`

auvipy
Oct 25, 2026
Maintainer

aviadr1 Oct 25, 2026
Author

aviadr1
Oct 25, 2026
Author