From 60385e452fcf0be006332a0ebbc0ca6038d86a3a Mon Sep 17 00:00:00 2001 From: William FH <13333726+hinthornw@users.noreply.github.com> Date: Sun, 15 Mar 2026 12:46:27 -0700 Subject: [PATCH] =?UTF-8?q?fix(checkpoint):=20don't=20add=20the=20task=20t?= =?UTF-8?q?o=20the=20checkpoint=20batch=20if=20it=20was=E2=80=A6=20(#7168)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cleaning up CI for #6701 ```md I'm trying to help a customer with some issues related to their checkpointing in postgres. They have some timeouts and retries around the langgraph checkpoint queries and I suspect the queue may be filling up with cancelled tasks. This PR adds some eager checks to not execute an operation if the future was already cancelled. This is to prevent the queue executing tasks that might have already timed out, which would otherwise cause more tasks to timeout due to the longer execution delay. Thank you for contributing to LangGraph! Follow these steps to mark your pull request as ready for review. **If any of these steps are not completed, your PR will not be considered for review.** ``` Co-authored-by: Conrad Ludgate --- libs/checkpoint/langgraph/store/base/batch.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/libs/checkpoint/langgraph/store/base/batch.py b/libs/checkpoint/langgraph/store/base/batch.py index 0a4a0eef7..64019d68f 100644 --- a/libs/checkpoint/langgraph/store/base/batch.py +++ b/libs/checkpoint/langgraph/store/base/batch.py @@ -328,6 +328,9 @@ async def _run( store: weakref.ReferenceType[BaseStore], ) -> None: while item := await aqueue.get(): + # don't run batch if the future is done (e.g. cancelled) + if item[0].done(): + continue # check if store is still alive if s := store(): try: @@ -335,6 +338,9 @@ async def _run( items = [item] try: while item := aqueue.get_nowait(): + # don't insert if the future is done (e.g. cancelled) + if item[0].done(): + continue items.append(item) except asyncio.QueueEmpty: pass