From ecde74a22208ebbec1352cd48cdf8a082bc2e4d6 Mon Sep 17 00:00:00 2001 From: yxstev Date: Fri, 10 Jul 2026 14:22:21 +0800 Subject: [PATCH 1/2] [fix] Clear storage data before releasing controller indexes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Previously, the clear operation released the storage unit index in the controller before freeing the underlying storage memory. A concurrent actor could acquire the released index, write new data, and then have that data freed by the trailing storage clear — causing silent data loss and missing-field errors on the index during long training runs. Fix by reversing the order: free storage memory first, then release the index back to the controller's reusable pool. Signed-off-by: yxstev --- transfer_queue/client.py | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/transfer_queue/client.py b/transfer_queue/client.py index 2c131a7..c117962 100644 --- a/transfer_queue/client.py +++ b/transfer_queue/client.py @@ -442,12 +442,12 @@ async def async_clear_partition(self, partition_id: str): ) return - # Clear the controller metadata - await self._clear_partition_in_controller(partition_id) - # Clear storage unit data await self.storage_manager.clear_data(metadata) + # Clear the controller metadata + await self._clear_partition_in_controller(partition_id) + logger.debug(f"[{self.client_id}]: Clear operation for partition_id {partition_id} completed.") except Exception as e: raise RuntimeError(f"Error in clear operation: {str(e)}") from e @@ -475,12 +475,13 @@ async def async_clear_samples(self, metadata: BatchMeta): if not self._controller: raise RuntimeError("No controller registered") - # Clear the controller metadata - await self._clear_meta_in_controller(metadata) - - # Clear storage unit data + # Clear storage unit data first, so released indexes are not reused + # before the underlying storage is actually freed. await self.storage_manager.clear_data(metadata) + # Clear the controller metadata (releases indexes to reusable pool) + await self._clear_meta_in_controller(metadata) + logger.debug(f"[{self.client_id}]: Clear operation for batch {metadata} completed.") except Exception as e: raise RuntimeError(f"Error in clear_samples operation: {str(e)}") from e From 5a9bb1acfac8b4af68acae6fd4e9f005b418814e Mon Sep 17 00:00:00 2001 From: yxstev Date: Fri, 10 Jul 2026 14:28:40 +0800 Subject: [PATCH 2/2] fix doc Signed-off-by: yxstev --- transfer_queue/client.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/transfer_queue/client.py b/transfer_queue/client.py index c117962..11caa3e 100644 --- a/transfer_queue/client.py +++ b/transfer_queue/client.py @@ -475,11 +475,10 @@ async def async_clear_samples(self, metadata: BatchMeta): if not self._controller: raise RuntimeError("No controller registered") - # Clear storage unit data first, so released indexes are not reused - # before the underlying storage is actually freed. + # Clear storage unit data await self.storage_manager.clear_data(metadata) - # Clear the controller metadata (releases indexes to reusable pool) + # Clear the controller metadata await self._clear_meta_in_controller(metadata) logger.debug(f"[{self.client_id}]: Clear operation for batch {metadata} completed.")