cleanup

libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py

@@ -32,7 +32,7 @@ Conn = _internal.Conn  # For backward compatibility
 class PostgresSaver(BasePostgresSaver):
     """Checkpointer that stores checkpoints in a Postgres database."""
 
-    lock: threading.RLock
+    lock: threading.Lock
 
     def __init__(
         self,
@@ -48,7 +48,7 @@ class PostgresSaver(BasePostgresSaver):
 
         self.conn = conn
         self.pipe = pipe
-        self.lock = threading.RLock()
+        self.lock = threading.Lock()
         self.supports_pipeline = Capabilities().has_pipeline()
 
     @classmethod
@@ -179,7 +179,7 @@ class PostgresSaver(BasePostgresSaver):
                             value["channel_values"],
                         )
             for value in values:
-                yield self._load_checkpoint_tuple(value)
+                yield self._load_checkpoint_tuple(value, cur)
 
     def get_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Get a checkpoint tuple from the database.
@@ -250,7 +250,7 @@ class PostgresSaver(BasePostgresSaver):
                         value["channel_values"],
                     )
 
-            return self._load_checkpoint_tuple(value)
+            return self._load_checkpoint_tuple(value, cur)
 
     def put(
         self,
@@ -430,22 +430,26 @@ class PostgresSaver(BasePostgresSaver):
                 with conn.cursor(binary=True, row_factory=dict_row) as cur:
                     yield cur
 
-    def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:
+    def _load_checkpoint_tuple(
+        self, value: DictRow, cur: Cursor[DictRow]
+    ) -> CheckpointTuple:
         """
         Convert a database row into a CheckpointTuple object.
 
         Args:
             value: A row from the database containing checkpoint data.
+            cur: The cursor used by the caller; reused for DeltaChannel
+                 reconstruction to avoid acquiring `self.lock` a second time.
 
         Returns:
             CheckpointTuple: A structured representation of the checkpoint,
             including its configuration, metadata, parent checkpoint (if any),
             and pending writes.
         """
-        from langgraph.checkpoint.base import DeltaChannelSentinel
+        from langgraph.checkpoint.base import DELTA_SENTINEL
 
         channel_values = self._load_blobs(value["channel_values"])
-        if any(isinstance(v, DeltaChannelSentinel) for v in channel_values.values()):
+        if any(v is DELTA_SENTINEL for v in channel_values.values()):
             cp_config = cast(
                 RunnableConfig,
                 {
@@ -456,8 +460,7 @@ class PostgresSaver(BasePostgresSaver):
                     }
                 },
             )
-            with self._cursor() as cur:
-                self._resolve_delta_channels(cp_config, channel_values, cur)
+            self._resolve_delta_channels(cp_config, channel_values, cur)
         return CheckpointTuple(
             {
                 "configurable": {

libs/checkpoint-postgres/langgraph/checkpoint/postgres/aio.py

@@ -8,12 +8,13 @@ from typing import Any
 
 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
+    DELTA_SENTINEL,
     WRITES_IDX_MAP,
     ChannelVersions,
     Checkpoint,
     CheckpointMetadata,
     CheckpointTuple,
-    DeltaChannelSentinel,
+    DeltaChannelWrites,
     get_checkpoint_id,
     get_serializable_checkpoint_metadata,
 )
@@ -169,7 +170,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
                             value["channel_values"],
                         )
             for value in values:
-                yield await self._load_checkpoint_tuple(value)
+                yield await self._load_checkpoint_tuple(value, cur)
 
     async def aget_tuple(self, config: RunnableConfig) -> CheckpointTuple | None:
         """Get a checkpoint tuple from the database asynchronously.
@@ -220,7 +221,7 @@ class AsyncPostgresSaver(BasePostgresSaver):
                         value["channel_values"],
                     )
 
-            return await self._load_checkpoint_tuple(value)
+            return await self._load_checkpoint_tuple(value, cur)
 
     async def aput(
         self,
@@ -444,12 +445,17 @@ class AsyncPostgresSaver(BasePostgresSaver):
                 thread_id, checkpoint_ns, checkpoint_id, channel, cur
             )
 
-    async def _load_checkpoint_tuple(self, value: DictRow) -> CheckpointTuple:
+    async def _load_checkpoint_tuple(
+        self, value: DictRow, cur: AsyncCursor[DictRow]
+    ) -> CheckpointTuple:
         """
         Convert a database row into a CheckpointTuple object.
 
         Args:
             value: A row from the database containing checkpoint data.
+            cur: The cursor used by the caller; reused for DeltaChannel
+                 reconstruction to avoid re-entering `self.lock`, which is
+                 `asyncio.Lock` and would deadlock.
 
         Returns:
             CheckpointTuple: A structured representation of the checkpoint,
@@ -464,17 +470,13 @@ class AsyncPostgresSaver(BasePostgresSaver):
         channel_values: dict[str, Any] = {}
         if blob_values:
             channel_values = self._load_blobs(blob_values)
-            delta_channels = [
-                ch
-                for ch, v in channel_values.items()
-                if isinstance(v, DeltaChannelSentinel)
-            ]
-            if delta_channels:
-                async with self._cursor() as cur:
-                    for channel in delta_channels:
-                        channel_values[channel] = await self._aget_channel_writes_cur(
+            for channel, v in channel_values.items():
+                if v is DELTA_SENTINEL:
+                    channel_values[channel] = DeltaChannelWrites(
+                        await self._aget_channel_writes_cur(
                             thread_id, checkpoint_ns, checkpoint_id, channel, cur
                         )
+                    )
 
         return CheckpointTuple(
             {

libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py

@@ -9,10 +9,11 @@ from typing import Any, cast
 
 from langchain_core.runnables import RunnableConfig
 from langgraph.checkpoint.base import (
+    DELTA_SENTINEL,
     WRITES_IDX_MAP,
     BaseCheckpointSaver,
     ChannelVersions,
-    DeltaChannelSentinel,
+    DeltaChannelWrites,
     get_checkpoint_id,
 )
 from langgraph.checkpoint.serde.types import TASKS
@@ -205,14 +206,16 @@ class BasePostgresSaver(BaseCheckpointSaver[str]):
         channel_values: dict[str, Any],
         cur: Any,
     ) -> None:
-        for channel, value in list(channel_values.items()):
-            if isinstance(value, DeltaChannelSentinel):
-                channel_values[channel] = self._get_channel_writes_cur(
-                    config["configurable"]["thread_id"],
-                    config["configurable"].get("checkpoint_ns", ""),
-                    config["configurable"]["checkpoint_id"],
-                    channel,
-                    cur,
+        for channel, value in channel_values.items():
+            if value is DELTA_SENTINEL:
+                channel_values[channel] = DeltaChannelWrites(
+                    self._get_channel_writes_cur(
+                        config["configurable"]["thread_id"],
+                        config["configurable"].get("checkpoint_ns", ""),
+                        config["configurable"]["checkpoint_id"],
+                        channel,
+                        cur,
+                    )
                 )
 
     def _get_channel_writes_cur(