From aeb9d7dfac6a69489b89f3bcc57928da95a4db7b Mon Sep 17 00:00:00 2001 From: "claude[bot]" <41898282+claude[bot]@users.noreply.github.com> Date: Sat, 21 Mar 2026 00:04:28 +0000 Subject: [PATCH 1/2] Fix race condition in topic/class description generation on concurrent uploads After each document reaches a terminal state (COMPLETED or FAILED), check whether all other documents in the same topic are also terminal before triggering generateTopicDescription and generateClassDescription. A distributed JobLock ensures exactly one API instance runs generation even when multiple completions race. Previously getSummarizationFuture triggered both description generation calls on every completing document, causing N parallel writes to the same Neo4j node. processFailedResult never triggered description regeneration at all. Changes: - DocumentRepository: add areAllDocumentsInTopicTerminal and areAllDocumentsInClassTerminal queries using Cypher all() predicate - ConvertingResultConsumer: inject JobLockRepository; extract triggerDescriptionGenerationIfAllTerminal which checks terminal state and acquires a per-topic/per-class JobLock before generating; call it asynchronously from both processSuccessfulResult and processFailedResult after persisting terminal status; remove now-unused topicRepository and classRepository fields Implements #108 Co-Authored-By: Claude Sonnet 4.6 --- .../api/queue/ConvertingResultConsumer.java | 137 ++++++++++++++---- .../api/repository/DocumentRepository.java | 38 +++++ 2 files changed, 150 insertions(+), 25 deletions(-) diff --git a/minerva-api/src/main/java/io/minerva/api/queue/ConvertingResultConsumer.java b/minerva-api/src/main/java/io/minerva/api/queue/ConvertingResultConsumer.java index de82cde9..0333aed9 100644 --- a/minerva-api/src/main/java/io/minerva/api/queue/ConvertingResultConsumer.java +++ b/minerva-api/src/main/java/io/minerva/api/queue/ConvertingResultConsumer.java @@ -142,16 +142,6 @@ public class ConvertingResultConsumer { */ private final DocumentSummarizationService documentSummarizationService; - /** - * Repository for loading {@link Topic} nodes by ID. - */ - private final TopicRepository topicRepository; - - /** - * Repository for loading {@link Clazz} nodes by ID. - */ - private final ClassRepository classRepository; - /** * Repository for saving {@link ChunkEmbedding} nodes and linking them to chunks. */ @@ -161,7 +151,18 @@ public class ConvertingResultConsumer { * Service for generating AI-suggested questions from a document. */ private final DocumentQuestionGenerationService documentQuestionGenerationService; - + + /** + * Repository used to acquire and release distributed locks that prevent duplicate topic/class description + * generation when multiple documents in the same topic or class complete concurrently. + */ + private final JobLockRepository jobLockRepository; + + /** + * Number of minutes a topic- or class-level description generation lock is held before it is considered stale. + */ + private static final int DESCRIPTION_LOCK_TIMEOUT_MINUTES = 10; + /** * Ensures the consumer group exists and starts the background consumer loop. Invoked automatically by Spring after * dependency injection. @@ -475,13 +476,23 @@ private void processSuccessfulResult(ConvertingResult result) { updatedDocument.setProcessedAt(Instant.now().atZone(ZoneOffset.UTC).toInstant()); updatedDocument.setStatus(DocumentStatus.COMPLETED); documentRepository.save(updatedDocument); - + + // Trigger topic/class description generation once all documents in the topic are terminal. + // Runs asynchronously so the consumer loop is not blocked while waiting for LLM responses. + UUID completedTopicId = updatedDocument.getTopicId(); + UUID completedClassId = updatedDocument.getClassId(); + if (completedTopicId != null && completedClassId != null) { + CompletableFuture.runAsync( + () -> triggerDescriptionGenerationIfAllTerminal(completedTopicId, completedClassId) + ); + } + String reverseKey = ConvertingTaskPublisher.DOCUMENT_TASK_PREFIX + document.getId().toString(); Object taskId = redisTemplate.opsForValue().get(reverseKey); if (taskId != null) { updatePipelineStep(taskId.toString(), "FINISHING", "COMPLETED"); } - + log.info("Document {} processed successfully.", document.getId()); } @@ -490,23 +501,88 @@ private CompletableFuture getSummarizationFuture(Document document, Conver () -> { try { documentSummarizationService.summarizeDocument(document.getId(), result.getContent()); - Topic topic = topicRepository.findById(document.getTopicId()) - .orElseThrow(() -> new RuntimeException( - "Topic not found: " + document.getTopicId())); - topicDescriptionGenerationService.generateTopicDescription(topic.getId()); - Clazz clazz = classRepository.findById(document.getClassId()) - .orElseThrow(() -> new RuntimeException( - "Class not found: " + document.getClassId())); - classDescriptionGenerationService.generateClassDescription(clazz.getId()); } catch (Exception e) { log.error( - "Summarization/description generation failed for document {}: {}", + "Summarization failed for document {}: {}", document.getId(), e.getMessage(), e ); } } ); } + + /** + * Triggers topic and class AI description generation if — and only if — every document in the same topic has + * reached a terminal state ({@code COMPLETED} or {@code FAILED}). + * + *

A {@link io.minerva.api.model.data.node.JobLock} is acquired for each level before generation begins, + * ensuring that exactly one API instance runs the generation even when multiple documents in the same topic finish + * concurrently. Each lock is released in a {@code finally} block. If the lock cannot be acquired (another instance + * is already generating), the call is silently skipped — the winning instance handles the work. + * + *

This method is called asynchronously (fire-and-forget) after the document status has been persisted, so + * the consumer loop is not blocked while waiting for LLM responses. + * + * @param topicId the ID of the topic whose document just reached a terminal state + * @param classId the ID of the class that owns the topic + */ + private void triggerDescriptionGenerationIfAllTerminal(UUID topicId, UUID classId) { + if (!documentRepository.areAllDocumentsInTopicTerminal(topicId)) { + log.debug( + "Skipping topic description generation for topic {} — not all documents are in terminal state", + topicId + ); + return; + } + + String topicLockName = "topic-desc:" + topicId; + boolean topicLockAcquired = false; + try { + topicLockAcquired = jobLockRepository.acquireLock(topicLockName, DESCRIPTION_LOCK_TIMEOUT_MINUTES); + if (topicLockAcquired) { + topicDescriptionGenerationService.generateTopicDescription(topicId); + } else { + log.debug( + "Topic description generation for topic {} already running on another instance, skipping", + topicId + ); + } + } catch (Exception e) { + log.error("Topic description generation failed for topic {}: {}", topicId, e.getMessage(), e); + } finally { + if (topicLockAcquired) { + jobLockRepository.releaseLock(topicLockName); + } + } + + if (!documentRepository.areAllDocumentsInClassTerminal(classId)) { + log.debug( + "Skipping class description generation for class {} — not all documents are in terminal state", + classId + ); + return; + } + + String classLockName = "class-desc:" + classId; + boolean classLockAcquired = false; + try { + classLockAcquired = jobLockRepository.acquireLock(classLockName, DESCRIPTION_LOCK_TIMEOUT_MINUTES); + if (classLockAcquired) { + classDescriptionGenerationService.generateClassDescription(classId); + } else { + log.debug( + "Class description generation for class {} already running on another instance, skipping", + classId + ); + } + } catch (Exception e) { + log.error("Class description generation failed for class {}: {}", classId, e.getMessage(), e); + } finally { + if (classLockAcquired) { + jobLockRepository.releaseLock(classLockName); + } + } + } private CompletableFuture getEmbeddingFuture(Document document, List chunks) { return CompletableFuture.runAsync( @@ -544,7 +620,8 @@ private CompletableFuture getQuestionFuture(Document document, ConvertingR } /** - * Handles a failed conversion result by marking the document as {@link DocumentStatus#FAILED}. + * Handles a failed conversion result by marking the document as {@link DocumentStatus#FAILED} and, if all other + * documents in the same topic have also reached a terminal state, triggering topic/class description generation. * * @param result the failed conversion result; must not be {@code null} */ @@ -557,10 +634,20 @@ private void processFailedResult(ConvertingResult result) { ); return; } - + document.setStatus(DocumentStatus.FAILED); documentRepository.save(document); - + + // Even when a document fails, description generation should still run once all documents in the topic + // have settled, so that partial content is not left without an AI description. + UUID failedTopicId = document.getTopicId(); + UUID failedClassId = document.getClassId(); + if (failedTopicId != null && failedClassId != null) { + CompletableFuture.runAsync( + () -> triggerDescriptionGenerationIfAllTerminal(failedTopicId, failedClassId) + ); + } + log.error("Document {} processing failed: {}", document.getId(), result.getError()); } diff --git a/minerva-api/src/main/java/io/minerva/api/repository/DocumentRepository.java b/minerva-api/src/main/java/io/minerva/api/repository/DocumentRepository.java index 9daa6be9..fc80e332 100644 --- a/minerva-api/src/main/java/io/minerva/api/repository/DocumentRepository.java +++ b/minerva-api/src/main/java/io/minerva/api/repository/DocumentRepository.java @@ -93,6 +93,44 @@ OPTIONAL MATCH (chunk)-[:HAS_CHUNK_EMBEDDING]->(chunkEmbedding:ChunkEmbedding) */ List findByUserId(@Param("userId") UUID userId); + /** + * Returns {@code true} when every document in the given topic has reached a terminal processing state + * ({@code COMPLETED} or {@code FAILED}) and at least one document exists. + * + *

Used by {@code ConvertingResultConsumer} to decide whether topic/class description generation + * should be triggered after a document reaches a terminal state. + * + * @param topicId the ID of the topic to check + * @return {@code true} if all documents are terminal and the topic is non-empty; {@code false} otherwise + */ + @Query( + """ + MATCH (d:Document {topicId: $topicId}) + WITH collect(d) AS docs + RETURN size(docs) > 0 AND all(doc IN docs WHERE doc.status IN ['COMPLETED', 'FAILED']) + """ + ) + boolean areAllDocumentsInTopicTerminal(@Param("topicId") UUID topicId); + + /** + * Returns {@code true} when every document across all topics in the given class has reached a terminal + * processing state ({@code COMPLETED} or {@code FAILED}) and at least one document exists. + * + *

Used by {@code ConvertingResultConsumer} to decide whether class description generation should be + * triggered after a topic's documents are all settled. + * + * @param classId the ID of the class to check + * @return {@code true} if all documents are terminal and the class is non-empty; {@code false} otherwise + */ + @Query( + """ + MATCH (d:Document {classId: $classId}) + WITH collect(d) AS docs + RETURN size(docs) > 0 AND all(doc IN docs WHERE doc.status IN ['COMPLETED', 'FAILED']) + """ + ) + boolean areAllDocumentsInClassTerminal(@Param("classId") UUID classId); + /** * Returns all documents for a topic together with their associated document summary (if any). * From 980b58ff8c942479011446f6426785725f17bf07 Mon Sep 17 00:00:00 2001 From: nilsknittel Date: Sat, 21 Mar 2026 01:40:12 +0100 Subject: [PATCH 2/2] Change method length to align with checkstyle --- .../api/queue/ConvertingResultConsumer.java | 79 +++++++++++-------- 1 file changed, 47 insertions(+), 32 deletions(-) diff --git a/minerva-api/src/main/java/io/minerva/api/queue/ConvertingResultConsumer.java b/minerva-api/src/main/java/io/minerva/api/queue/ConvertingResultConsumer.java index 0333aed9..e3171275 100644 --- a/minerva-api/src/main/java/io/minerva/api/queue/ConvertingResultConsumer.java +++ b/minerva-api/src/main/java/io/minerva/api/queue/ConvertingResultConsumer.java @@ -1,10 +1,16 @@ package io.minerva.api.queue; -import io.minerva.api.model.data.node.*; +import io.minerva.api.model.data.node.Chunk; +import io.minerva.api.model.data.node.ChunkEmbedding; +import io.minerva.api.model.data.node.Document; +import io.minerva.api.model.data.node.DocumentPartition; import io.minerva.api.model.dto.document.DocumentChunkingDto; import io.minerva.api.model.status.DocumentStatus; import io.minerva.api.queue.model.ConvertingResult; -import io.minerva.api.repository.*; +import io.minerva.api.repository.ChunkEmbeddingRepository; +import io.minerva.api.repository.ChunkRepository; +import io.minerva.api.repository.DocumentRepository; +import io.minerva.api.repository.JobLockRepository; import io.minerva.api.service.*; import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; @@ -151,18 +157,18 @@ public class ConvertingResultConsumer { * Service for generating AI-suggested questions from a document. */ private final DocumentQuestionGenerationService documentQuestionGenerationService; - + /** * Repository used to acquire and release distributed locks that prevent duplicate topic/class description * generation when multiple documents in the same topic or class complete concurrently. */ private final JobLockRepository jobLockRepository; - + /** * Number of minutes a topic- or class-level description generation lock is held before it is considered stale. */ private static final int DESCRIPTION_LOCK_TIMEOUT_MINUTES = 10; - + /** * Ensures the consumer group exists and starts the background consumer loop. Invoked automatically by Spring after * dependency injection. @@ -433,23 +439,7 @@ private void processSuccessfulResult(ConvertingResult result) { ); document.setParsedUrl(parsedUrl); } - document.setChunkCount(result.getNumChunks()); - document.setImageCount(result.getNumImages()); - document.setPageCount(result.getNumPages()); - document.setTableCount(result.getNumTables()); - document.setFormulaCount(result.getNumFormulas()); - - // Persist YouTube-specific metadata when available - if (result.getVideoTitle() != null && !result.getVideoTitle().isBlank()) { - document.setTitle(result.getVideoTitle()); - } - if (result.getThumbnailUrl() != null && !result.getThumbnailUrl().isBlank()) { - document.setThumbnailUrl(result.getThumbnailUrl()); - } - if (result.getVideoDuration() != null) { - document.setVideoDuration(result.getVideoDuration()); - } - + setDocumentMetadata(document, result); documentRepository.save(document); // Convert and filter chunks @@ -476,7 +466,7 @@ private void processSuccessfulResult(ConvertingResult result) { updatedDocument.setProcessedAt(Instant.now().atZone(ZoneOffset.UTC).toInstant()); updatedDocument.setStatus(DocumentStatus.COMPLETED); documentRepository.save(updatedDocument); - + // Trigger topic/class description generation once all documents in the topic are terminal. // Runs asynchronously so the consumer loop is not blocked while waiting for LLM responses. UUID completedTopicId = updatedDocument.getTopicId(); @@ -486,16 +476,41 @@ private void processSuccessfulResult(ConvertingResult result) { () -> triggerDescriptionGenerationIfAllTerminal(completedTopicId, completedClassId) ); } - + String reverseKey = ConvertingTaskPublisher.DOCUMENT_TASK_PREFIX + document.getId().toString(); Object taskId = redisTemplate.opsForValue().get(reverseKey); if (taskId != null) { updatePipelineStep(taskId.toString(), "FINISHING", "COMPLETED"); } - + log.info("Document {} processed successfully.", document.getId()); } + /** + * Enrich document node with metadata from {@link ConvertingResult} + * + * @param document The {@link Document} Node + * @param result {@link ConvertingResult} with Metadat + */ + private void setDocumentMetadata(Document document, ConvertingResult result) { + document.setChunkCount(result.getNumChunks()); + document.setImageCount(result.getNumImages()); + document.setPageCount(result.getNumPages()); + document.setTableCount(result.getNumTables()); + document.setFormulaCount(result.getNumFormulas()); + + // Persist YouTube-specific metadata when available + if (result.getVideoTitle() != null && !result.getVideoTitle().isBlank()) { + document.setTitle(result.getVideoTitle()); + } + if (result.getThumbnailUrl() != null && !result.getThumbnailUrl().isBlank()) { + document.setThumbnailUrl(result.getThumbnailUrl()); + } + if (result.getVideoDuration() != null) { + document.setVideoDuration(result.getVideoDuration()); + } + } + private CompletableFuture getSummarizationFuture(Document document, ConvertingResult result) { return CompletableFuture.runAsync( () -> { @@ -510,7 +525,7 @@ private CompletableFuture getSummarizationFuture(Document document, Conver } ); } - + /** * Triggers topic and class AI description generation if — and only if — every document in the same topic has * reached a terminal state ({@code COMPLETED} or {@code FAILED}). @@ -534,7 +549,7 @@ private void triggerDescriptionGenerationIfAllTerminal(UUID topicId, UUID classI ); return; } - + String topicLockName = "topic-desc:" + topicId; boolean topicLockAcquired = false; try { @@ -554,7 +569,7 @@ private void triggerDescriptionGenerationIfAllTerminal(UUID topicId, UUID classI jobLockRepository.releaseLock(topicLockName); } } - + if (!documentRepository.areAllDocumentsInClassTerminal(classId)) { log.debug( "Skipping class description generation for class {} — not all documents are in terminal state", @@ -562,7 +577,7 @@ private void triggerDescriptionGenerationIfAllTerminal(UUID topicId, UUID classI ); return; } - + String classLockName = "class-desc:" + classId; boolean classLockAcquired = false; try { @@ -634,10 +649,10 @@ private void processFailedResult(ConvertingResult result) { ); return; } - + document.setStatus(DocumentStatus.FAILED); documentRepository.save(document); - + // Even when a document fails, description generation should still run once all documents in the topic // have settled, so that partial content is not left without an AI description. UUID failedTopicId = document.getTopicId(); @@ -647,7 +662,7 @@ private void processFailedResult(ConvertingResult result) { () -> triggerDescriptionGenerationIfAllTerminal(failedTopicId, failedClassId) ); } - + log.error("Document {} processing failed: {}", document.getId(), result.getError()); }