{"record":{"id":"1bdeea6403231896","repo":"harry0703/MoneyPrinterTurbo","slug":"task-queue-is-full-please-try-again-later","errorCode":null,"errorMessage":"task queue is full, please try again later","messagePattern":"task queue is full, please try again later","errorType":"exception","errorClass":"TaskQueueFullError","httpStatus":429,"severity":"warning","filePath":"app/controllers/manager/base_manager.py","lineNumber":46,"sourceCode":"                # 在线程启动前先预占并发名额。原实现在线程内部递增，连续请求\n                # 可能都在子线程获得锁之前看到 current_tasks=0，从而突破并发\n                # 上限。启动失败时回滚名额，让后续请求仍可正常调度。\n                self.current_tasks += 1\n                try:\n                    self.execute_task(func, *args, **kwargs)\n                except Exception:\n                    self.current_tasks -= 1\n                    raise\n            else:\n                queue_size = self.queue_size()\n                # 并发数已满时才进入排队。队列必须有上限，否则匿名接口可以持续\n                # 堆积任务对象和请求参数，最终造成内存耗尽或第三方 API 成本失控。\n                if queue_size >= self.max_queued_tasks:\n                    logger.warning(\n                        f\"reject task: {func.__name__}, queue_size: {queue_size}, \"\n                        f\"max_queued_tasks: {self.max_queued_tasks}\"\n                    )\n                    raise TaskQueueFullError(\"task queue is full, please try again later\")\n\n                logger.info(\n                    f\"enqueue task: {func.__name__}, current_tasks: {self.current_tasks}, \"\n                    f\"queue_size: {queue_size}\"\n                )\n                self.enqueue({\"func\": func, \"args\": args, \"kwargs\": kwargs})\n\n    def execute_task(self, func: Callable, *args: Any, **kwargs: Any):\n        thread = threading.Thread(\n            target=self.run_task, args=(func, *args), kwargs=kwargs\n        )\n        thread.start()\n\n    def run_task(self, func: Callable, *args: Any, **kwargs: Any):\n        try:\n            func(*args, **kwargs)  # call the function here, passing *args and **kwargs.\n        finally:\n            self.task_done()","sourceCodeStart":28,"sourceCodeEnd":64,"githubUrl":"https://github.com/harry0703/MoneyPrinterTurbo/blob/1f9f19c2021a68d04df228f33e9099a0c947f6f8/app/controllers/manager/base_manager.py#L28-L64","documentation":"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'.","triggerScenarios":"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.","commonSituations":"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.","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."],"exampleFix":"# before\nfor i in range(50):\n    resp = requests.post(f\"{base}/api/v1/videos\", json=payload, headers=h)\n    resp.raise_for_status()\n\n# after\nimport time, random\nfor i in range(50):\n    for attempt in range(5):\n        resp = requests.post(f\"{base}/api/v1/videos\", json=payload, headers=h)\n        if resp.status_code != 429:\n            break\n        time.sleep((2 ** attempt) + random.random())\n    resp.raise_for_status()","handlingStrategy":"retry","validationCode":"# before submitting, check current load\ninfo = requests.get(f\"{base}/api/v1/tasks\", headers=h).json()\n# if visible running+queued counts are near your known limits, delay submission","typeGuard":"def is_queue_full_response(resp) -> bool:\n    return resp.status_code == 429 and \"queue is full\" in resp.text","tryCatchPattern":"for attempt in range(6):\n    resp = requests.post(url, json=body, headers=h)\n    if resp.status_code != 429:\n        break\n    time.sleep(min(60, 2 ** attempt + random.random()))\nelse:\n    raise RuntimeError(\"task queue still full after backoff\")","preventionTips":["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."],"tags":["backpressure","queue","http-429","capacity"],"backgroundTag":null,"analyzedSha":"1f9f19c2021a68d04df228f33e9099a0c947f6f8","analyzedAt":"2026-08-14T19:41:05.568Z","schemaVersion":2},"datasetVersion":"2026-08-15T17:31:12.345Z"}