[fix] Complete KV storage async cleanup

[ji-huazhong]

Background

• KVStorageManager.get_data() and clear_data() invoked synchronous backend calls directly from async methods, blocking the caller's event loop. put_data() used the loop's default executor, which was outside the manager's lifecycle.
• TensorDict reconstruction still occupied the caller's event loop after get(), which could stall unrelated coroutines in a Ray worker or async actor.
• The executor finalizer captured None before the lazy executor was created, so it could not release the actual thread pool. KVStorageManager.close() also did not close backend-owned resources such as MooncakeStoreClient.

What this changes

• Run synchronous put, get, and clear operations on a manager-owned storage executor, and keep get plus TensorDict reconstruction off the caller's event loop.
• Keep storage I/O and reconstruction executors separate so concurrent get operations cannot deadlock through nested submissions to the same pool.
• Add a default no-op StorageKVClient.close() lifecycle hook and make KVStorageManager.close() idempotently drain both executors, close the backend client, and then release the existing notify and ZMQ resources.
• Remove the ineffective finalizer and reject new executor work after close.
• Add regression coverage for non-blocking async dispatch, reconstruction thread placement, executor shutdown, backend cleanup, and idempotent close.

Tests

• python -m compileall -q transfer_queue tutorial tests
• python -m ruff check transfer_queue/storage/clients/base.py transfer_queue/storage/managers/base.py tests/test_kv_storage_manager.py
• python -m pytest -q tests/test_kv_storage_manager.py tests/test_storage_client_factory.py Result: 11 passed, 2 skipped.
• python -m pytest -q tests/e2e/test_kv_interface_e2e.py::TestRayWorkerKVInterfaceE2E Result: 2 passed for sync and async Ray remote-worker paths.
• python -m pytest -q --ignore=tests/test_yuanrong_storage_client_e2e.py Result: 571 passed, 10 skipped.
• The unfiltered suite reached 571 passed and 10 skipped, with 8 setup errors in the pre-existing Yuanrong mock test because the verl environment has no yr package and yuanrong_client therefore exposes no datasystem attribute.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 3642f7eb29

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".

[chatgpt-codex-connector]

Keep GDR get work on the selected CUDA device

When MooncakeStore_GDR is used after the process selects a non-default GPU before tq.init(), this now runs MooncakeStoreClient.get() on a fresh executor thread. That GDR path uses torch.cuda.current_device() when lazily initializing/unpacking the staging buffer, and CUDA current device is thread-local, so the worker can fall back to cuda:0 instead of the device the caller selected, causing buffers/results to be allocated on the wrong GPU or failing on multi-GPU processes.

AGENTS.md reference: AGENTS.md:L17-L18

Useful? React with 👍 / 👎.