From 4ee15b163d0d27ff5db14e769f7a33ca65dbf0bf Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Mon, 5 Jan 2026 13:22:46 +0000 Subject: [PATCH] - Storage processor ignore data on the queue which is for a deleted collection --- .../trustgraph/base/collection_config_handler.py | 7 ++++++- .../storage/doc_embeddings/pinecone/write.py | 11 ++++++++++- .../trustgraph/storage/doc_embeddings/qdrant/write.py | 11 ++++++++++- .../storage/graph_embeddings/pinecone/write.py | 11 ++++++++++- .../storage/graph_embeddings/qdrant/write.py | 11 ++++++++++- 5 files changed, 46 insertions(+), 5 deletions(-) diff --git a/trustgraph-base/trustgraph/base/collection_config_handler.py b/trustgraph-base/trustgraph/base/collection_config_handler.py index 97130d1e..8c1af822 100644 --- a/trustgraph-base/trustgraph/base/collection_config_handler.py +++ b/trustgraph-base/trustgraph/base/collection_config_handler.py @@ -77,10 +77,15 @@ class CollectionConfigHandler: for user, collection in deleted_collections: logger.info(f"Collection deleted: {user}/{collection}") try: - await self.delete_collection(user, collection) + # Remove from known_collections FIRST to immediately reject new writes + # This eliminates race condition with worker threads del self.known_collections[(user, collection)] + # Physical deletion happens after - worker threads already rejecting writes + await self.delete_collection(user, collection) except Exception as e: logger.error(f"Error deleting collection {user}/{collection}: {e}", exc_info=True) + # If physical deletion failed, should we re-add to known_collections? + # For now, keep it removed - collection is logically deleted per config logger.debug(f"Collection config processing complete. Known collections: {len(self.known_collections)}") diff --git a/trustgraph-flow/trustgraph/storage/doc_embeddings/pinecone/write.py b/trustgraph-flow/trustgraph/storage/doc_embeddings/pinecone/write.py index 846855a4..6d1b23ba 100644 --- a/trustgraph-flow/trustgraph/storage/doc_embeddings/pinecone/write.py +++ b/trustgraph-flow/trustgraph/storage/doc_embeddings/pinecone/write.py @@ -90,6 +90,15 @@ class Processor(CollectionConfigHandler, DocumentEmbeddingsStoreService): async def store_document_embeddings(self, message): + # Validate collection exists in config before processing + if not self.collection_exists(message.metadata.user, message.metadata.collection): + logger.warning( + f"Collection {message.metadata.collection} for user {message.metadata.user} " + f"does not exist in config (likely deleted while data was in-flight). " + f"Dropping message." + ) + return + for emb in message.chunks: if emb.chunk is None or emb.chunk == b"": continue @@ -105,7 +114,7 @@ class Processor(CollectionConfigHandler, DocumentEmbeddingsStoreService): f"d-{message.metadata.user}-{message.metadata.collection}-{dim}" ) - # Lazily create index if it doesn't exist + # Lazily create index if it doesn't exist (but only if authorized in config) if not self.pinecone.has_index(index_name): logger.info(f"Lazily creating Pinecone index {index_name} with dimension {dim}") self.create_index(index_name, dim) diff --git a/trustgraph-flow/trustgraph/storage/doc_embeddings/qdrant/write.py b/trustgraph-flow/trustgraph/storage/doc_embeddings/qdrant/write.py index 923ade10..edfa8aa9 100644 --- a/trustgraph-flow/trustgraph/storage/doc_embeddings/qdrant/write.py +++ b/trustgraph-flow/trustgraph/storage/doc_embeddings/qdrant/write.py @@ -41,6 +41,15 @@ class Processor(CollectionConfigHandler, DocumentEmbeddingsStoreService): async def store_document_embeddings(self, message): + # Validate collection exists in config before processing + if not self.collection_exists(message.metadata.user, message.metadata.collection): + logger.warning( + f"Collection {message.metadata.collection} for user {message.metadata.user} " + f"does not exist in config (likely deleted while data was in-flight). " + f"Dropping message." + ) + return + for emb in message.chunks: chunk = emb.chunk.decode("utf-8") @@ -54,7 +63,7 @@ class Processor(CollectionConfigHandler, DocumentEmbeddingsStoreService): f"d_{message.metadata.user}_{message.metadata.collection}_{dim}" ) - # Lazily create collection if it doesn't exist + # Lazily create collection if it doesn't exist (but only if authorized in config) if not self.qdrant.collection_exists(collection): logger.info(f"Lazily creating Qdrant collection {collection} with dimension {dim}") self.qdrant.create_collection( diff --git a/trustgraph-flow/trustgraph/storage/graph_embeddings/pinecone/write.py b/trustgraph-flow/trustgraph/storage/graph_embeddings/pinecone/write.py index 9fe38720..0bee6ceb 100755 --- a/trustgraph-flow/trustgraph/storage/graph_embeddings/pinecone/write.py +++ b/trustgraph-flow/trustgraph/storage/graph_embeddings/pinecone/write.py @@ -90,6 +90,15 @@ class Processor(CollectionConfigHandler, GraphEmbeddingsStoreService): async def store_graph_embeddings(self, message): + # Validate collection exists in config before processing + if not self.collection_exists(message.metadata.user, message.metadata.collection): + logger.warning( + f"Collection {message.metadata.collection} for user {message.metadata.user} " + f"does not exist in config (likely deleted while data was in-flight). " + f"Dropping message." + ) + return + for entity in message.entities: if entity.entity.value == "" or entity.entity.value is None: @@ -103,7 +112,7 @@ class Processor(CollectionConfigHandler, GraphEmbeddingsStoreService): f"t-{message.metadata.user}-{message.metadata.collection}-{dim}" ) - # Lazily create index if it doesn't exist + # Lazily create index if it doesn't exist (but only if authorized in config) if not self.pinecone.has_index(index_name): logger.info(f"Lazily creating Pinecone index {index_name} with dimension {dim}") self.create_index(index_name, dim) diff --git a/trustgraph-flow/trustgraph/storage/graph_embeddings/qdrant/write.py b/trustgraph-flow/trustgraph/storage/graph_embeddings/qdrant/write.py index 39127473..e3c2b6bc 100755 --- a/trustgraph-flow/trustgraph/storage/graph_embeddings/qdrant/write.py +++ b/trustgraph-flow/trustgraph/storage/graph_embeddings/qdrant/write.py @@ -41,6 +41,15 @@ class Processor(CollectionConfigHandler, GraphEmbeddingsStoreService): async def store_graph_embeddings(self, message): + # Validate collection exists in config before processing + if not self.collection_exists(message.metadata.user, message.metadata.collection): + logger.warning( + f"Collection {message.metadata.collection} for user {message.metadata.user} " + f"does not exist in config (likely deleted while data was in-flight). " + f"Dropping message." + ) + return + for entity in message.entities: if entity.entity.value == "" or entity.entity.value is None: return @@ -53,7 +62,7 @@ class Processor(CollectionConfigHandler, GraphEmbeddingsStoreService): f"t_{message.metadata.user}_{message.metadata.collection}_{dim}" ) - # Lazily create collection if it doesn't exist + # Lazily create collection if it doesn't exist (but only if authorized in config) if not self.qdrant.collection_exists(collection): logger.info(f"Lazily creating Qdrant collection {collection} with dimension {dim}") self.qdrant.create_collection(