What causes "task queue is full, please try again later" in harry0703/MoneyPrinterTurbo?

Treat the 429 as retryable: back off (e.g. exponential with jitter, starting ~5-10s) and resubmit once queued tasks drain.

harry0703/MoneyPrinterTurbo · warning · TaskQueueFullError

task queue is full, please try again later

Error message

task queue is full, please try again later

What it means

TaskQueueFullError is raised by the decorated submit path in app/controllers/manager/base_manager.py when concurrency is saturated (current_tasks >= max_concurrent_tasks) and the pending queue has already reached max_queued_tasks. The bounded queue exists deliberately: without it, anonymous endpoints could pile up task objects and request parameters until memory is exhausted or third-party API costs run away. It surfaces to HTTP as a 429 with the message 'task queue is full, please try again later'.

Source

Thrown at app/controllers/manager/base_manager.py:46

                # 在线程启动前先预占并发名额。原实现在线程内部递增，连续请求
                # 可能都在子线程获得锁之前看到 current_tasks=0，从而突破并发
                # 上限。启动失败时回滚名额，让后续请求仍可正常调度。
                self.current_tasks += 1
                try:
                    self.execute_task(func, *args, **kwargs)
                except Exception:
                    self.current_tasks -= 1
                    raise
            else:
                queue_size = self.queue_size()
                # 并发数已满时才进入排队。队列必须有上限，否则匿名接口可以持续
                # 堆积任务对象和请求参数，最终造成内存耗尽或第三方 API 成本失控。
                if queue_size >= self.max_queued_tasks:
                    logger.warning(
                        f"reject task: {func.__name__}, queue_size: {queue_size}, "
                        f"max_queued_tasks: {self.max_queued_tasks}"
                    )
                    raise TaskQueueFullError("task queue is full, please try again later")

                logger.info(
                    f"enqueue task: {func.__name__}, current_tasks: {self.current_tasks}, "
                    f"queue_size: {queue_size}"
                )
                self.enqueue({"func": func, "args": args, "kwargs": kwargs})

    def execute_task(self, func: Callable, *args: Any, **kwargs: Any):
        thread = threading.Thread(
            target=self.run_task, args=(func, *args), kwargs=kwargs
        )
        thread.start()

    def run_task(self, func: Callable, *args: Any, **kwargs: Any):
        try:
            func(*args, **kwargs)  # call the function here, passing *args and **kwargs.
        finally:
            self.task_done()

View on GitHub (pinned to 1f9f19c202)

Solutions

Treat the 429 as retryable: back off (e.g. exponential with jitter, starting ~5-10s) and resubmit once queued tasks drain.
If this load is expected, raise max_concurrent_tasks and/or max_queued_tasks in the InMemoryTaskManager configuration and restart the service.
Inspect running tasks (GET /api/v1/tasks) to find stuck tasks; delete or cancel them so workers free up and the queue drains.
Rate-limit the submitting client to spread arrivals instead of bursting.

Example fix

# before
for i in range(50):
    resp = requests.post(f"{base}/api/v1/videos", json=payload, headers=h)
    resp.raise_for_status()

# after
import time, random
for i in range(50):
    for attempt in range(5):
        resp = requests.post(f"{base}/api/v1/videos", json=payload, headers=h)
        if resp.status_code != 429:
            break
        time.sleep((2 ** attempt) + random.random())
    resp.raise_for_status()

Defensive patterns

Strategy: retry

Validate before calling

# before submitting, check current load
info = requests.get(f"{base}/api/v1/tasks", headers=h).json()
# if visible running+queued counts are near your known limits, delay submission

Type guard

def is_queue_full_response(resp) -> bool:
    return resp.status_code == 429 and "queue is full" in resp.text

Try / catch

for attempt in range(6):
    resp = requests.post(url, json=body, headers=h)
    if resp.status_code != 429:
        break
    time.sleep(min(60, 2 ** attempt + random.random()))
else:
    raise RuntimeError("task queue still full after backoff")

Prevention

Always implement exponential backoff with jitter on task-creation endpoints.
Size max_queued_tasks to your worst expected burst, not your average load.
Monitor queue depth and alert before saturation so submitters slow down first.

When it happens

Trigger: POST /api/v1/videos repeatedly while generation tasks are long-running and max_concurrent_tasks workers are busy AND queue_size() >= max_queued_tasks; burst load from an automated client with no backoff; slow upstream video-generation APIs stalling task drain.

Common situations: Default max_queued_tasks too low for the expected burst size; a stuck third-party API call holding all worker slots so the queue never drains; load tests or batch scripts firing many requests in parallel without retry logic.

Related errors

{request_id}: {str(e)}

AI-assisted analysis of harry0703/MoneyPrinterTurbo@1f9f19c202 (2026-08-14). Data as JSON: /api/errors/1bdeea6403231896. Report an issue: GitHub.