apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-23 20:21:03 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

- Storage processor ignore data on the queue which is for a deleted

Browse source 
collection
This commit is contained in:
Cyber MacGeddon 2026-01-05 13:22:46 +00:00
parent 142e13dc7a
commit 4ee15b163d
5 changed files with 46 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

7
trustgraph-base/trustgraph/base/collection_config_handler.py
View file

@ -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)}")

11
trustgraph-flow/trustgraph/storage/doc_embeddings/pinecone/write.py
View file

@ -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)

11
trustgraph-flow/trustgraph/storage/doc_embeddings/qdrant/write.py
View file

@ -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(

11
trustgraph-flow/trustgraph/storage/graph_embeddings/pinecone/write.py
View file

@ -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)

11
trustgraph-flow/trustgraph/storage/graph_embeddings/qdrant/write.py
View file

@ -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(