Конвейеры django-dramatiq запускают только первый этап

Я пытаюсь использовать django-dramatiq для запуска конвейера из нескольких этапов, каждый из которых определен как dramatiq Actor, используя метод pipeline(<stages>).run(), но он запускает только первый этап/Actor и не пытается запустить другие этапы.

Я определил несколько урезанных поддельных актеров, чтобы проиллюстрировать проблему:

import dramatiq

@dramatiq.actor
def fake_extract(process_pk, *args, **kwargs):
    print(f"fake_extract: Process PK= {process_pk} Running extract on {kwargs['fits_file']}")

@dramatiq.actor
def fake_astromfit(process_pk, *args, **kwargs):
    print(f"fake_astromfit: Process PK= {process_pk} Astrometric fit on {kwargs['ldac_catalog']}, updating {kwargs['fits_file']}")

@dramatiq.actor
def fake_zeropoint(process_pk, *args, **kwargs):
    print(f"fake_zeropoint: Process PK= {process_pk} ZP determination on {kwargs['ldac_catalog']} with {kwargs['desired_catalog']} ref catalog")

Затем я определил этапы и построил конвейер:

import os
from dramatiq import pipeline
from test_dramatiq.dramatiq_tests import fake_extract, fake_astromfit, fake_zeropoint


fits_filepath = '/foo/bar.fits'
fits_file = os.path.basename(fits_filepath)

steps = [{
            'name'   : 'proc-extract',
            'runner' : fake_extract,
            'inputs' : {'fits_file':fits_filepath,
                       'datadir': os.path.join(dataroot, temp_dir)}
        },
        {
            'name'   : 'proc-astromfit',
            'runner' : fake_astromfit,
            'inputs' : {'fits_file' : fits_filepath,
                        'ldac_catalog' : os.path.join(dataroot, temp_dir, fits_file.replace('e91.fits', 'e91_ldac.fits')),
                        'datadir' : os.path.join(dataroot, temp_dir)
                        }
        },
        {
            'name'   : 'proc-zeropoint',
            'runner' : fake_zeropoint,
            'inputs' : {'ldac_catalog' : os.path.join(dataroot, temp_dir, fits_file.replace('e91.fits', 'e92_ldac.fits')),
                        'datadir' : os.path.join(dataroot, temp_dir),
                        'desired_catalog' : 'PS1'
                        }
        }]

pipes = []
for step_num, step in enumerate(steps):
    inputs = step['inputs']
    print(f"  Performing pipeline step {step['name']}")
    pk = 1234+step_num
    pipes.append(step['runner'].message_with_options(args=[pk,], kwargs=inputs, pipe_ignore=True))
pipeline(pipes).run()

Запуск внутри ipython с обычными dramatiq эти, кажется, работают нормально и все этапы выполняются:

fake_extract: Process PK= 1234 Running extract on /foo/bar.fits
fake_astromfit: Process PK= 1235 Astrometric fit on /foo/Temp_cvc2/bar.fits, updating /foo/bar.fits
fake_zeropoint: Process PK= 1236 ZP determination on /foo/Temp_cvc2/bar.fits with PS1 ref catalog

Однако определение их в модуле, импортированном django-dramatiq через файл settings.py проекта Django, и определение конвейера, как указано выше, в python manage.py shell и с помощью бегуна python manage.py rundramatiq запускает только первый этап/Actor:

fake_extract: Process PK= 1234 Running extract on /foo/bar.fits

и он никогда не выполняет другие этапы... Есть идеи, что здесь происходит и почему многоступенчатые конвейеры не работают под django-dramatiq ?

DRAMATIQ_BROKER = {
    'BROKER': 'dramatiq.brokers.redis.RedisBroker',
    'OPTIONS': {
        'url': f'redis://{REDIS_HOSTNAME}:6379',
    },
    'MIDDLEWARE': [
        'dramatiq.middleware.AgeLimit',
        'dramatiq.middleware.TimeLimit',
        'dramatiq.middleware.Callbacks',
        'dramatiq.middleware.Retries',
        'django_dramatiq.middleware.DbConnectionsMiddleware',
    ]
}