Maximum recursion depth exceeded while encoding a JSON object for a long pipeline

[synweap15]

What OS are you using?

Ubuntu 20.04

What version of Dramatiq are you using?

1.9.0

What did you do?

I've created a pipeline with about 1400 tasks that I wanted to be processed in order:

    pipeline_tasks = []
    for entry in entries:
        pipeline_tasks.append(
            reprocess_source_image.message_with_options(
                kwargs={"source_image_id": entry.id}, pipe_ignore=True
            )
        )

    pipeline(pipeline_tasks).run()

What did you expect would happen?

Pipeline goes through.

What happened?

Pipeline failed with the following exception:

builtins:RecursionError: maximum recursion depth exceeded while encoding a JSON object

File "/home/deploy/venv/lib/python3.8/site-packages/dramatiq/composition.py", line 116, in run
File "/home/deploy/venv/lib/python3.8/site-packages/dramatiq/brokers/redis.py", line 182, in enqueue
File "/home/deploy/venv/lib/python3.8/site-packages/dramatiq/message.py", line 101, in encode
File "/home/deploy/venv/lib/python3.8/site-packages/dramatiq/encoder.py", line 49, in encode
File "/usr/lib/python3.8/json/__init__.py", line 234, in dumps
File "/usr/lib/python3.8/json/encoder.py", line 199, in encode
File "/usr/lib/python3.8/json/encoder.py", line 257, in iterencode

I think it could be handled more gracefully with checking sys.getrecursionlimit() before creating a pipeline, or ideally have the pipeline creation redesigned a bit.

[sadilet]

@synweap15 Hi, could you please tell me, how did you solve it eventually?

[synweap15]

@sadilet I've taken the short path and restructured the single long pipeline into n smaller pipelines. Then the order condition is not satisfied, but it was fine enough for my case