From 604257e33d17880be7ad4cc8d04290c9cfe7b9c2 Mon Sep 17 00:00:00 2001 From: Kyle Barron Date: Tue, 29 Sep 2026 18:37:54 -0400 Subject: [PATCH 1/2] fix: detach from the interpreter while creating a list stream `list()` created the stream while still attached. `MemoryStore` takes its read lock at that point, and a concurrent `put()` or `delete()` releases the replaced Python buffer while holding the write lock, which needs to re-attach. On GIL builds the lister holds the GIL while waiting for the lock; on free-threaded builds it blocks a pending stop-the-world. Either way both threads wait on each other forever. Create the stream inside `py.detach`, as `list_with_delimiter()` already does. The regression test runs in a subprocess because on failure the stuck lister holds the GIL and would hang the test process itself. Closes #785 Co-Authored-By: Claude Opus 5.5 --- obstore/src/list.rs | 15 ++++++++----- tests/test_list.py | 51 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 61 insertions(+), 5 deletions(-) diff --git a/obstore/src/list.rs b/obstore/src/list.rs index 253787c2..4fcc19f0 100644 --- a/obstore/src/list.rs +++ b/obstore/src/list.rs @@ -371,11 +371,16 @@ pub(crate) fn list( let store = store.into_inner().clone(); let prefix = prefix.map(|s| s.into()); - let stream = if let Some(offset) = offset { - store.list_with_offset(prefix.as_ref(), offset.as_ref()) - } else { - store.list(prefix.as_ref()) - }; + // Detach because creating the stream may block: MemoryStore takes its lock here, and + // a concurrent writer can need to attach to release a Python buffer while holding it. + // https://github.com/developmentseed/obstore/issues/785 + let stream = py.detach(|| { + if let Some(offset) = offset { + store.list_with_offset(prefix.as_ref(), offset.as_ref()) + } else { + store.list(prefix.as_ref()) + } + }); Ok(PyListStream::new(stream, chunk_size, return_arrow)) } diff --git a/tests/test_list.py b/tests/test_list.py index dcec45a2..7377ee53 100644 --- a/tests/test_list.py +++ b/tests/test_list.py @@ -1,3 +1,7 @@ +import subprocess +import sys +import textwrap + import pytest from arro3.core import RecordBatch, Table @@ -159,3 +163,50 @@ async def test_list_with_delimiter_async(): assert objects.num_rows == 2 assert objects["path"][0].as_py() == "a/file1.txt" assert objects["path"][1].as_py() == "a/file2.txt" + + +def test_list_concurrent_with_put_does_not_deadlock(): + # https://github.com/developmentseed/obstore/issues/785 + # + # Each put replaces an earlier put's Python-backed buffer, and MemoryStore releases + # that buffer while holding its write lock. list() must not wait on the store's + # lock while attached to the interpreter. This runs in a subprocess because on + # failure the stuck lister holds the GIL, which would hang the test process too. + script = textwrap.dedent( + """ + import gc + import threading + + import obstore as obs + from obstore.store import MemoryStore + + store = MemoryStore() + for i in range(50): + obs.put(store, f"k/{i}", b"x" * 1024) + + def writer(): + for n in range(1_000): + obs.put(store, f"k/{n % 50}", bytes(1024)) + + def lister(): + for _ in range(1_000): + obs.list(store) + + def offset_lister(): + for _ in range(1_000): + obs.list(store, offset="k/0") + + def collector(): + # Free-threaded builds need a stop-the-world to hit the deadlock. + for _ in range(100): + gc.collect() + + targets = (writer, writer, lister, offset_lister, collector) + threads = [threading.Thread(target=target) for target in targets] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + """, + ) + subprocess.run([sys.executable, "-c", script], check=True, timeout=60) # noqa: S603 From beec7990b023ec59fdea1d5033212e009a7c0578 Mon Sep 17 00:00:00 2001 From: Kyle Barron Date: Tue, 29 Sep 2026 19:14:24 -0400 Subject: [PATCH 2/2] chore: drop the list deadlock regression test and comment Co-Authored-By: Claude Opus 5.5 --- obstore/src/list.rs | 3 --- tests/test_list.py | 51 --------------------------------------------- 2 files changed, 54 deletions(-) diff --git a/obstore/src/list.rs b/obstore/src/list.rs index 4fcc19f0..69486554 100644 --- a/obstore/src/list.rs +++ b/obstore/src/list.rs @@ -371,9 +371,6 @@ pub(crate) fn list( let store = store.into_inner().clone(); let prefix = prefix.map(|s| s.into()); - // Detach because creating the stream may block: MemoryStore takes its lock here, and - // a concurrent writer can need to attach to release a Python buffer while holding it. - // https://github.com/developmentseed/obstore/issues/785 let stream = py.detach(|| { if let Some(offset) = offset { store.list_with_offset(prefix.as_ref(), offset.as_ref()) diff --git a/tests/test_list.py b/tests/test_list.py index 7377ee53..dcec45a2 100644 --- a/tests/test_list.py +++ b/tests/test_list.py @@ -1,7 +1,3 @@ -import subprocess -import sys -import textwrap - import pytest from arro3.core import RecordBatch, Table @@ -163,50 +159,3 @@ async def test_list_with_delimiter_async(): assert objects.num_rows == 2 assert objects["path"][0].as_py() == "a/file1.txt" assert objects["path"][1].as_py() == "a/file2.txt" - - -def test_list_concurrent_with_put_does_not_deadlock(): - # https://github.com/developmentseed/obstore/issues/785 - # - # Each put replaces an earlier put's Python-backed buffer, and MemoryStore releases - # that buffer while holding its write lock. list() must not wait on the store's - # lock while attached to the interpreter. This runs in a subprocess because on - # failure the stuck lister holds the GIL, which would hang the test process too. - script = textwrap.dedent( - """ - import gc - import threading - - import obstore as obs - from obstore.store import MemoryStore - - store = MemoryStore() - for i in range(50): - obs.put(store, f"k/{i}", b"x" * 1024) - - def writer(): - for n in range(1_000): - obs.put(store, f"k/{n % 50}", bytes(1024)) - - def lister(): - for _ in range(1_000): - obs.list(store) - - def offset_lister(): - for _ in range(1_000): - obs.list(store, offset="k/0") - - def collector(): - # Free-threaded builds need a stop-the-world to hit the deadlock. - for _ in range(100): - gc.collect() - - targets = (writer, writer, lister, offset_lister, collector) - threads = [threading.Thread(target=target) for target in targets] - for thread in threads: - thread.start() - for thread in threads: - thread.join() - """, - ) - subprocess.run([sys.executable, "-c", script], check=True, timeout=60) # noqa: S603