langchain-ai/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-07-15 12:35:43 -04:00
Code Issues 314 Packages Projects Releases 17 Wiki Activity

msgpack is not thread-safe, which can causing corrupt checkpoints #557

New Issue
Closed
opened 2026-02-20 17:40:42 -05:00 by yindo · 8 comments
yindo commented 2026-02-20 17:40:42 -05:00
Owner
Copy Link
Copy Source

Originally created by @marcammann on GitHub (Apr 3, 2025).

Checked other resources

  • This is a bug, not a usage question. For questions, please use GitHub Discussions.
  • I added a clear and detailed title that summarizes the issue.
  • I read what a minimal reproducible example is (https://stackoverflow.com/help/minimal-reproducible-example).
  • I included a self-contained, minimal example that demonstrates the issue INCLUDING all the relevant imports. The code run AS IS to reproduce the issue.

Example Code

import asyncio

import msgpack
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
from msgpack.fallback import Unpacker

serde = JsonPlusSerializer()


def _pack_repr(x):
    binary_representation = ' '.join(format(byte, '02x') for byte in x)
    return binary_representation


def _pack(value, i):
    packed = serde.dumps_typed(value)
    return packed


def _add_message(value: dict, i):
    messages = value['nested']['messages']
    messages.append(HumanMessage(content="Baz!"))


async def _test(value, i: int):
    """
    Spawns two threads, one to pack the value and one to add a message to the value.
    """
    pack_thread = asyncio.to_thread(
        _pack,
        value,
        i,
    )

    add_thread = asyncio.to_thread(
        _add_message,
        value,
        i,
    )

    _, packed = await asyncio.gather(
        add_thread,
        pack_thread,
    )

    try:
        serde.loads_typed(packed)
    except Exception as e:
        print(f"Error in thread {i}: {e}")


async def _run():
    messages = [HumanMessage(content=f"Message #{i}") for i in range(1000)]
    value = {"messages": messages, "nested": {
        "messages": messages
    }}

    tasks = []
    for i in range(100):
        cp = value.copy()
        tasks.append(_test(cp, i))

        # print(f"Packed value: {_pack_repr(packed[1])[:100]}")
    await asyncio.gather(*tasks)


if __name__ == "__main__":
    asyncio.run(_run())

Error Message and Stack Trace (if applicable)

`Error in thread 48: unpack(b) received extra data.`


  File ".venv/lib/python3.12/site-packages/langgraph/checkpoint/serde/jsonplus.py", line 209, in loads_typed
    return msgpack.unpackb(
           ^^^^^^^^^^^^^^^^
  File "msgpack/_unpacker.pyx", line 194, in msgpack._cmsgpack.unpackb

Description

We're using the Postgres Checkpointer in our project and we find ourselves with a large number of messages, which lead to the discovery of a threading issue in the async variant of the checkpointer library.

(We are filtering messages before sending them to an LLM using placeholders, so not to worry there)

The issues seems to stem from msgpack itself not being thread safe (and by extension, serde). With large checkpoint objects, the packing steps takes just enough time for multi threading issues to be present frequently. While the JsonPlusSerializer does a good job at creating multiple encoders to ensure that each thread has its own encoder, it doesn't address the issue of the checkpoint object being mutated while the packing is happening.

This happens because the copy of the checkpoint object is only shallow:

copy = checkpoint.copy()
        ...

async with self._cursor(pipeline=True) as cur:
    await cur.executemany(
        self.UPSERT_CHECKPOINT_BLOBS_SQL,
        await asyncio.to_thread(
            self._dump_blobs,
            thread_id,
            checkpoint_ns,
            copy.pop("channel_values"),  # type: ignore[misc]
            new_versions,
        ),
    )

What seems to happen is that msgpack determines the length of an array (messages in our case) but then messages is mutated in another thread, leading to the header of the packed value being incorrect for the amount of elements in the list.

The example code above is a little contrived to illustrate the issue.

Ultimately this leads to a corrupted checkpoint, as loading the checkpoint throws an exception as the unpacking error isn't caught anywhere. That then leads to our users not being able to invoke the graph at all.

I'm happy to work on a PR to address the issue, I'm however unsure how you want to address it, ideally.

  • protected sections?
  • deep copy of the checkpoint?

System Info

System Information

OS: Darwin
OS Version: Darwin Kernel Version 24.3.0: Thu Jan 2 20:22:58 PST 2025; root:xnu-11215.81.4~3/RELEASE_ARM64_T8132
Python Version: 3.12.7 (main, Oct 16 2024, 07:12:08) [Clang 18.1.8 ]

Package Information

langchain_core: 0.3.35
langchain: 0.3.17
langchain_community: 0.3.0
langsmith: 0.1.144
langchain_openai: 0.3.6
langchain_postgres: 0.0.12
langchain_text_splitters: 0.3.5
langchainhub: 0.1.20
langgraph_sdk: 0.1.51

Optional packages not installed

langserve

Other Dependencies

aiohttp: 3.11.7
async-timeout: Installed. No version info available.
dataclasses-json: 0.6.7
httpx: 0.27.2
jsonpatch<2.0,>=1.33: Installed. No version info available.
langchain-core<1.0.0,>=0.3.35: Installed. No version info available.
langsmith<0.4,>=0.1.125: Installed. No version info available.
numpy: 1.26.4
openai<2.0.0,>=1.58.1: Installed. No version info available.
orjson: 3.10.11
packaging: 24.2
packaging<25,>=23.2: Installed. No version info available.
pgvector: 0.2.5
psycopg: 3.2.3
psycopg-pool: 3.2.4
pydantic: 2.10.0
pydantic-settings: 2.6.1
pydantic<3.0.0,>=2.5.2;: Installed. No version info available.
pydantic<3.0.0,>=2.7.4;: Installed. No version info available.
PyYAML: 6.0.2
PyYAML>=5.3: Installed. No version info available.
requests: 2.32.3
requests-toolbelt: 1.0.0
sqlalchemy: 2.0.36
SQLAlchemy: 2.0.36
tenacity: 8.5.0
tenacity!=8.4.0,<10.0.0,>=8.1.0: Installed. No version info available.
tiktoken<1,>=0.7: Installed. No version info available.
types-requests: 2.32.0.20241016
typing-extensions>=4.7: Installed. No version info available.

Originally created by @marcammann on GitHub (Apr 3, 2025). ### Checked other resources - [x] This is a bug, not a usage question. For questions, please use GitHub Discussions. - [x] I added a clear and detailed title that summarizes the issue. - [x] I read what a minimal reproducible example is (https://stackoverflow.com/help/minimal-reproducible-example). - [x] I included a self-contained, minimal example that demonstrates the issue INCLUDING all the relevant imports. The code run AS IS to reproduce the issue. ### Example Code ```python import asyncio import msgpack from langchain_core.messages import HumanMessage from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer from msgpack.fallback import Unpacker serde = JsonPlusSerializer() def _pack_repr(x): binary_representation = ' '.join(format(byte, '02x') for byte in x) return binary_representation def _pack(value, i): packed = serde.dumps_typed(value) return packed def _add_message(value: dict, i): messages = value['nested']['messages'] messages.append(HumanMessage(content="Baz!")) async def _test(value, i: int): """ Spawns two threads, one to pack the value and one to add a message to the value. """ pack_thread = asyncio.to_thread( _pack, value, i, ) add_thread = asyncio.to_thread( _add_message, value, i, ) _, packed = await asyncio.gather( add_thread, pack_thread, ) try: serde.loads_typed(packed) except Exception as e: print(f"Error in thread {i}: {e}") async def _run(): messages = [HumanMessage(content=f"Message #{i}") for i in range(1000)] value = {"messages": messages, "nested": { "messages": messages }} tasks = [] for i in range(100): cp = value.copy() tasks.append(_test(cp, i)) # print(f"Packed value: {_pack_repr(packed[1])[:100]}") await asyncio.gather(*tasks) if __name__ == "__main__": asyncio.run(_run()) ``` ### Error Message and Stack Trace (if applicable) ```shell `Error in thread 48: unpack(b) received extra data.` File ".venv/lib/python3.12/site-packages/langgraph/checkpoint/serde/jsonplus.py", line 209, in loads_typed return msgpack.unpackb( ^^^^^^^^^^^^^^^^ File "msgpack/_unpacker.pyx", line 194, in msgpack._cmsgpack.unpackb ``` ### Description We're using the Postgres Checkpointer in our project and we find ourselves with a large number of messages, which lead to the discovery of a threading issue in the async variant of the checkpointer library. (We are filtering messages before sending them to an LLM using placeholders, so not to worry there) The issues seems to stem from `msgpack` itself not being thread safe (and by extension, `serde`). With large checkpoint objects, the packing steps takes just enough time for multi threading issues to be present frequently. While the JsonPlusSerializer does a good job at creating multiple encoders to ensure that each thread has its own encoder, it doesn't address the issue of the checkpoint object being mutated while the packing is happening. This happens because the copy of the checkpoint object is only shallow: ``` copy = checkpoint.copy() ... async with self._cursor(pipeline=True) as cur: await cur.executemany( self.UPSERT_CHECKPOINT_BLOBS_SQL, await asyncio.to_thread( self._dump_blobs, thread_id, checkpoint_ns, copy.pop("channel_values"), # type: ignore[misc] new_versions, ), ) ``` What seems to happen is that `msgpack` determines the length of an array (messages in our case) but then messages is mutated in another thread, leading to the header of the packed value being incorrect for the amount of elements in the list. The example code above is a little contrived to illustrate the issue. Ultimately this leads to a corrupted checkpoint, as loading the checkpoint throws an exception as the unpacking error isn't caught anywhere. That then leads to our users not being able to invoke the graph at all. I'm happy to work on a PR to address the issue, I'm however unsure how you want to address it, ideally. - protected sections? - deep copy of the checkpoint? ### System Info System Information ------------------ > OS: Darwin > OS Version: Darwin Kernel Version 24.3.0: Thu Jan 2 20:22:58 PST 2025; root:xnu-11215.81.4~3/RELEASE_ARM64_T8132 > Python Version: 3.12.7 (main, Oct 16 2024, 07:12:08) [Clang 18.1.8 ] Package Information ------------------- > langchain_core: 0.3.35 > langchain: 0.3.17 > langchain_community: 0.3.0 > langsmith: 0.1.144 > langchain_openai: 0.3.6 > langchain_postgres: 0.0.12 > langchain_text_splitters: 0.3.5 > langchainhub: 0.1.20 > langgraph_sdk: 0.1.51 Optional packages not installed ------------------------------- > langserve Other Dependencies ------------------ > aiohttp: 3.11.7 > async-timeout: Installed. No version info available. > dataclasses-json: 0.6.7 > httpx: 0.27.2 > jsonpatch<2.0,>=1.33: Installed. No version info available. > langchain-core<1.0.0,>=0.3.35: Installed. No version info available. > langsmith<0.4,>=0.1.125: Installed. No version info available. > numpy: 1.26.4 > openai<2.0.0,>=1.58.1: Installed. No version info available. > orjson: 3.10.11 > packaging: 24.2 > packaging<25,>=23.2: Installed. No version info available. > pgvector: 0.2.5 > psycopg: 3.2.3 > psycopg-pool: 3.2.4 > pydantic: 2.10.0 > pydantic-settings: 2.6.1 > pydantic<3.0.0,>=2.5.2;: Installed. No version info available. > pydantic<3.0.0,>=2.7.4;: Installed. No version info available. > PyYAML: 6.0.2 > PyYAML>=5.3: Installed. No version info available. > requests: 2.32.3 > requests-toolbelt: 1.0.0 > sqlalchemy: 2.0.36 > SQLAlchemy: 2.0.36 > tenacity: 8.5.0 > tenacity!=8.4.0,<10.0.0,>=8.1.0: Installed. No version info available. > tiktoken<1,>=0.7: Installed. No version info available. > types-requests: 2.32.0.20241016 > typing-extensions>=4.7: Installed. No version info available.
yindo closed this issue 2026-02-20 17:40:42 -05:00
yindo commented 2026-02-20 17:40:43 -05:00
Author
Owner
Copy Link
Copy Source

@hinthornw commented on GitHub (Apr 3, 2025):

Hey thanks for reporting - will investigate. Out of curiosity why are you mutating state in LangGraph?

Updates are meant to be done via responses from the node. Mutations are mostly an anti-pattern. Deep-copying the state every time would incur an unnecessary performance penalty for most folks who aren't trying to mutate things so I'm on the fence about going that direction

@hinthornw commented on GitHub (Apr 3, 2025): Hey thanks for reporting - will investigate. Out of curiosity why are you mutating state in LangGraph? Updates are meant to be done via responses from the node. Mutations are mostly an anti-pattern. Deep-copying the state every time would incur an unnecessary performance penalty for most folks who aren't trying to mutate things so I'm on the fence about going that direction
yindo commented 2026-02-20 17:40:43 -05:00
Author
Owner
Copy Link
Copy Source

@marcammann commented on GitHub (Apr 3, 2025):

We're not, I think the issue stems from multiple writes/puts happening at the same time and/or when nodes are adding messages in quick succession, since the actual reference of the messages never updates. So the nodes write to the same list of messages that's currently being "packed".

@marcammann commented on GitHub (Apr 3, 2025): We're not, I think the issue stems from multiple writes/puts happening at the same time and/or when nodes are adding messages in quick succession, since the actual reference of the `messages` never updates. So the nodes write to the same list of messages that's currently being "packed".
yindo commented 2026-02-20 17:40:43 -05:00
Author
Owner
Copy Link
Copy Source

@hinthornw commented on GitHub (Apr 3, 2025):

Ah got it! My mistake. Do you have an example graph that would reproduce this?

@hinthornw commented on GitHub (Apr 3, 2025): Ah got it! My mistake. Do you have an example graph that would reproduce this?
yindo commented 2026-02-20 17:40:43 -05:00
Author
Owner
Copy Link
Copy Source

@marcammann commented on GitHub (Apr 3, 2025):

Our main graph is a bit large (with various sub graphs) to cut down into an example. I'm not sure how easy this is to reproduce in a real graph as it's a pretty rare issue and only occurs with large checkpoint objects? I was thinking about how I would even write a test for this (both on our end and in LangGraph)

I could potentially try to just do a simple two to three node graph and see how that reacts, will see if I find some time tomorrow.

@marcammann commented on GitHub (Apr 3, 2025): Our main graph is a bit large (with various sub graphs) to cut down into an example. I'm not sure how easy this is to reproduce in a real graph as it's a pretty rare issue and only occurs with large checkpoint objects? I was thinking about how I would even write a test for this (both on our end and in LangGraph) I could potentially try to just do a simple two to three node graph and see how that reacts, will see if I find some time tomorrow.
yindo commented 2026-02-20 17:40:43 -05:00
Author
Owner
Copy Link
Copy Source

@hinthornw commented on GitHub (Apr 4, 2025):

Is this still occurring on more recent versions? We replaced msgpack with ormsgpack recently.

I don't expect the behavior to be that different though and will investigate more

@hinthornw commented on GitHub (Apr 4, 2025): Is this still occurring on more recent versions? We replaced `msgpack` with `ormsgpack` recently. I don't expect the behavior to be that different though and will investigate more
yindo commented 2026-02-20 17:40:43 -05:00
Author
Owner
Copy Link
Copy Source

@hinthornw commented on GitHub (Apr 4, 2025):

Hm yeah actually I can't reproduce this error on anything after langgraph-checkpoint==2.0.22, which is when we switched serializer libs.

@hinthornw commented on GitHub (Apr 4, 2025): Hm yeah actually I can't reproduce this error on anything after `langgraph-checkpoint==2.0.22`, which is when we switched serializer libs.
yindo commented 2026-02-20 17:40:43 -05:00
Author
Owner
Copy Link
Copy Source

@hinthornw commented on GitHub (Apr 4, 2025):

@marcammann I believe this is already fixed if you upgrade langgraph-checkpoint please let us know if you run into any further issues!

@hinthornw commented on GitHub (Apr 4, 2025): @marcammann I believe this is already fixed if you upgrade `langgraph-checkpoint` please let us know if you run into any further issues!
yindo commented 2026-02-20 17:40:43 -05:00
Author
Owner
Copy Link
Copy Source

@marcammann commented on GitHub (Apr 4, 2025):

Sounds great! Will report back. Thank you!

@marcammann commented on GitHub (Apr 4, 2025): Sounds great! Will report back. Thank you!
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
agents
awaiting author revision
bug
change
chat models
checkpointer
ci
code quality
code quality
dependencies
documentation
drawing
duplicate
enhancement
good first issue
help wanted
invalid
maintainer
meta
missing information
not maintained by langchain
open-swe-auto
open-swe-max
open-swe-max-auto
pending
performance
pull-request
Mirrored from GitHub Pull Request
 question
requires design
sdk
server
stale
studio
tests
type checking
usage question
v0.5
v1.1
No Label
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langchain-ai/langgraph#557
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?