Skip to content

fix(ingestion): prune stale vectors when a changed file is guard-rejected #237

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
detail-app wants to merge 1 commit into
base: dev
Choose a base branch
from
Open

fix(ingestion): prune stale vectors when a changed file is guard-rejected #237

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 44 additions & 8 deletions .../java/com/williamcallahan/javachat/service/ingestion/LocalDocsFileIngestionProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -297,7 +297,10 @@ private FilePreparation prepare(Path root, Path file, Map<Path, String> ingestio
GuardDecision guardDecision = contentGuard.evaluate(new GuardInput(bodyText, parsedDocument));
if (!guardDecision.acceptable()) {
String rejectionReason = guardDecision.rejectionReason();
return deferred(() -> quarantineRejectedFile(file, rejectionReason));
final boolean replacementRequired = requiresFullReindex;
final MarkerContext rejectedMarkerContext = markerContext;
return deferred(
() -> quarantineRejectedFile(rejectedMarkerContext, replacementRequired, rejectionReason));
}
}
}
Expand Down Expand Up @@ -731,22 +734,55 @@ private static boolean isNavigationOnlyDocument(org.jsoup.nodes.Document parsedD
return navigationFrameset || interactiveApiReferenceShell;
}

private LocalDocsFileOutcome quarantineRejectedFile(Path file, String rejectionReason) {
private LocalDocsFileOutcome quarantineRejectedFile(
MarkerContext markerContext, boolean requiresFullReindex, String rejectionReason) {
Path file = markerContext.file();
String contentGuardPhase;
String contentGuardDetails;
try {
var quarantineService = fileContentServices.quarantine();
IngestionQuarantineService.QuarantineResult quarantineCopy = quarantineService.quarantine(file);
INDEXING_LOG.warn("[INDEXING] Content guard rejected file and copied it to quarantine");
return LocalDocsFileOutcome.failedFile(new IngestionLocalFailure(
file.toString(),
"content-guard",
"quarantine copy " + quarantineCopy.quarantined() + ": " + rejectionReason));
contentGuardPhase = "content-guard";
contentGuardDetails = "quarantine copy " + quarantineCopy.quarantined() + ": " + rejectionReason;
} catch (IOException quarantineException) {
log.warn(
"Failed to quarantine invalid content (exception type: {})",
quarantineException.getClass().getSimpleName());
return LocalDocsFileOutcome.failedFile(
failureFactory.failure(file, "quarantine-write", quarantineException));
contentGuardPhase = "quarantine-write";
contentGuardDetails = failureFactory
.failure(file, "quarantine-write", quarantineException)
.details();
}
if (requiresFullReindex) {
try {
storage.hybridVector().deleteByUrl(markerContext.collectionKind(), markerContext.url());
ingestedFilePruneService.pruneObsoleteLocalStateAfterReplacement(
markerContext.url(),
markerContext.priorIngestionRecord().orElse(null),
List.of());
} catch (IOException pruneException) {
return LocalDocsFileOutcome.failedFile(failureFactory.failure(file, "prune-local", pruneException));
} catch (RuntimeException pruneException) {
return LocalDocsFileOutcome.failedFile(failureFactory.failure(file, "prune-runtime", pruneException));
}
try {
markFileIngested(
markerContext.url(),
new FileIngestionRecord(
markerContext.fileSizeBytes(),
markerContext.lastModifiedMillis(),
markerContext.ingestionFingerprint(),
LOCAL_DOCS_EXTRACTION_SEMANTICS_VERSION,
markerContext.collectionName(),
List.of()));
} catch (RuntimeException markerTransitionException) {
return LocalDocsFileOutcome.failedFile(
failureFactory.failure(file, "marker-transition", markerTransitionException));
}
}
return LocalDocsFileOutcome.failedFile(
new IngestionLocalFailure(file.toString(), contentGuardPhase, contentGuardDetails));
}

private LocalDocsFileOutcome markPreviouslyIngestedFile(
Expand Down
171 changes: 171 additions & 0 deletions ...a/com/williamcallahan/javachat/service/ingestion/LocalDocsFileIngestionProcessorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -468,6 +468,177 @@ void shouldStopBeforeLaterFileWhenContentIsRejected(@TempDir Path temporaryDirec
.processAndStoreChunks(anyString(), anyString(), anyString(), anyString());
}

@Test
void shouldNotLeaveStaleVectorsWhenPreviouslyIngestedFileIsGuardRejected(@TempDir Path temporaryDirectory)
throws IOException {
DocumentationSource documentationSource =
DocsSourceRegistry.documentationSources().getFirst();
Path selectedDocumentationRoot =
temporaryDirectory.resolve("corpus").resolve(documentationSource.relativeMirrorPath());
Files.createDirectories(selectedDocumentationRoot);
Path rejectedFile = selectedDocumentationRoot.resolve("rejected.html");
Files.writeString(
rejectedFile,
"<html><head><title>404 Not Found</title></head><body><h1>404 Not Found</h1></body></html>",
StandardCharsets.UTF_8);
String expectedUrl = DocsSourceRegistry.resolveMirroredPath(selectedDocumentationRoot, rejectedFile)
.orElseThrow();
FileIngestionRecord priorIngestionRecord = new FileIngestionRecord(
Files.size(rejectedFile),
Files.getLastModifiedTime(rejectedFile).toMillis(),
"prior-good-fingerprint",
LocalDocsFileIngestionProcessor.LOCAL_DOCS_EXTRACTION_SEMANTICS_VERSION,
"documentation",
List.of("prior-chunk-hash"));

LocalDocsIngestionFixture ingestionFixture = new LocalDocsIngestionFixture();
when(ingestionFixture.fileIngestionMarkerStore.readFileIngestionRecord(expectedUrl))
.thenReturn(Optional.of(priorIngestionRecord));
when(ingestionFixture.hybridVectorService.resolveCollectionName(any())).thenReturn("documentation");
when(ingestionFixture.quarantineService.quarantine(rejectedFile))
.thenReturn(new IngestionQuarantineService.QuarantineResult(
rejectedFile, temporaryDirectory.resolve("quarantine/rejected.html")));

LocalDocsFileOutcome outcome =
ingestionFixture.ingestionProcessor().process(selectedDocumentationRoot, rejectedFile);

assertFalse(outcome.processed());
assertEquals("content-guard", outcome.failure().orElseThrow().phase());
verify(ingestionFixture.quarantineService).quarantine(rejectedFile);
verify(ingestionFixture.hybridVectorService).deleteByUrl(any(QdrantCollectionKind.class), eq(expectedUrl));
verify(ingestionFixture.ingestedFilePruneService)
.pruneObsoleteLocalStateAfterReplacement(expectedUrl, priorIngestionRecord, List.of());
ArgumentCaptor<FileIngestionRecord> markerCaptor = ArgumentCaptor.forClass(FileIngestionRecord.class);
verify(ingestionFixture.fileIngestionMarkerStore).markFileIngested(eq(expectedUrl), markerCaptor.capture());
FileIngestionRecord rejectedIngestionRecord = markerCaptor.getValue();
assertTrue(rejectedIngestionRecord.chunkHashes().isEmpty());
verify(ingestionFixture.chunkProcessingService, never())
.processAndStoreChunks(anyString(), anyString(), anyString(), anyString());
}

@Test
void shouldNotLeaveStaleVectorsWhenQuarantineWriteFailsForPreviouslyIngestedFile(@TempDir Path temporaryDirectory)
throws IOException {
DocumentationSource documentationSource =
DocsSourceRegistry.documentationSources().getFirst();
Path selectedDocumentationRoot =
temporaryDirectory.resolve("corpus").resolve(documentationSource.relativeMirrorPath());
Files.createDirectories(selectedDocumentationRoot);
Path rejectedFile = selectedDocumentationRoot.resolve("rejected.html");
Files.writeString(
rejectedFile,
"<html><head><title>404 Not Found</title></head><body><h1>404 Not Found</h1></body></html>",
StandardCharsets.UTF_8);
String expectedUrl = DocsSourceRegistry.resolveMirroredPath(selectedDocumentationRoot, rejectedFile)
.orElseThrow();
FileIngestionRecord priorIngestionRecord = new FileIngestionRecord(
Files.size(rejectedFile),
Files.getLastModifiedTime(rejectedFile).toMillis(),
"prior-good-fingerprint",
LocalDocsFileIngestionProcessor.LOCAL_DOCS_EXTRACTION_SEMANTICS_VERSION,
"documentation",
List.of("prior-chunk-hash"));

LocalDocsIngestionFixture ingestionFixture = new LocalDocsIngestionFixture();
when(ingestionFixture.fileIngestionMarkerStore.readFileIngestionRecord(expectedUrl))
.thenReturn(Optional.of(priorIngestionRecord));
when(ingestionFixture.hybridVectorService.resolveCollectionName(any())).thenReturn("documentation");
doThrow(new IOException("quarantine storage unavailable"))
.when(ingestionFixture.quarantineService)
.quarantine(rejectedFile);

LocalDocsFileOutcome outcome =
ingestionFixture.ingestionProcessor().process(selectedDocumentationRoot, rejectedFile);

assertFalse(outcome.processed());
assertEquals("quarantine-write", outcome.failure().orElseThrow().phase());
verify(ingestionFixture.quarantineService).quarantine(rejectedFile);
verify(ingestionFixture.hybridVectorService).deleteByUrl(any(QdrantCollectionKind.class), eq(expectedUrl));
verify(ingestionFixture.ingestedFilePruneService)
.pruneObsoleteLocalStateAfterReplacement(expectedUrl, priorIngestionRecord, List.of());
ArgumentCaptor<FileIngestionRecord> markerCaptor = ArgumentCaptor.forClass(FileIngestionRecord.class);
verify(ingestionFixture.fileIngestionMarkerStore).markFileIngested(eq(expectedUrl), markerCaptor.capture());
FileIngestionRecord rejectedIngestionRecord = markerCaptor.getValue();
assertTrue(rejectedIngestionRecord.chunkHashes().isEmpty());
verify(ingestionFixture.chunkProcessingService, never())
.processAndStoreChunks(anyString(), anyString(), anyString(), anyString());
}

@Test
void shouldBeIdempotentOnNextRunAfterGuardRejectedPreviouslyIngestedFile(@TempDir Path temporaryDirectory)
throws IOException {
DocumentationSource documentationSource =
DocsSourceRegistry.documentationSources().getFirst();
Path selectedDocumentationRoot =
temporaryDirectory.resolve("corpus").resolve(documentationSource.relativeMirrorPath());
Files.createDirectories(selectedDocumentationRoot);
Path rejectedFile = selectedDocumentationRoot.resolve("rejected.html");
Files.writeString(
rejectedFile,
"<html><head><title>404 Not Found</title></head><body><h1>404 Not Found</h1></body></html>",
StandardCharsets.UTF_8);
String expectedUrl = DocsSourceRegistry.resolveMirroredPath(selectedDocumentationRoot, rejectedFile)
.orElseThrow();
FileIngestionRecord priorIngestionRecord = new FileIngestionRecord(
Files.size(rejectedFile),
Files.getLastModifiedTime(rejectedFile).toMillis(),
"prior-good-fingerprint",
LocalDocsFileIngestionProcessor.LOCAL_DOCS_EXTRACTION_SEMANTICS_VERSION,
"documentation",
List.of("prior-chunk-hash"));

LocalDocsIngestionFixture ingestionFixture = new LocalDocsIngestionFixture();
Map<String, FileIngestionRecord> markerStore = new HashMap<>();
markerStore.put(expectedUrl, priorIngestionRecord);
when(ingestionFixture.fileIngestionMarkerStore.readFileIngestionRecord(anyString()))
.thenAnswer(invocation -> Optional.ofNullable(markerStore.get(invocation.getArgument(0))));
doAnswer(invocation -> {
markerStore.put(invocation.getArgument(0), invocation.getArgument(1));
return null;
})
.when(ingestionFixture.fileIngestionMarkerStore)
.markFileIngested(anyString(), any(FileIngestionRecord.class));
when(ingestionFixture.hybridVectorService.resolveCollectionName(any())).thenReturn("documentation");
when(ingestionFixture.quarantineService.quarantine(rejectedFile))
.thenReturn(new IngestionQuarantineService.QuarantineResult(
rejectedFile, temporaryDirectory.resolve("quarantine/rejected.html")));

LocalDocsFileOutcome firstOutcome =
ingestionFixture.ingestionProcessor().process(selectedDocumentationRoot, rejectedFile);

assertFalse(firstOutcome.processed());
assertEquals("content-guard", firstOutcome.failure().orElseThrow().phase());
verify(ingestionFixture.hybridVectorService).deleteByUrl(any(QdrantCollectionKind.class), eq(expectedUrl));
verify(ingestionFixture.ingestedFilePruneService)
.pruneObsoleteLocalStateAfterReplacement(expectedUrl, priorIngestionRecord, List.of());
verify(ingestionFixture.fileIngestionMarkerStore)
.markFileIngested(eq(expectedUrl), any(FileIngestionRecord.class));

clearInvocations(
ingestionFixture.chunkProcessingService,
ingestionFixture.hybridVectorService,
ingestionFixture.fileIngestionMarkerStore,
ingestionFixture.ingestedFilePruneService,
ingestionFixture.quarantineService);
when(ingestionFixture.hybridVectorService.hasExactPointIdsForUrl(
any(QdrantCollectionKind.class), eq(expectedUrl), eq(List.of())))
.thenReturn(true);

LocalDocsFileOutcome repeatedOutcome =
ingestionFixture.ingestionProcessor().process(selectedDocumentationRoot, rejectedFile);

assertFalse(repeatedOutcome.processed());
assertTrue(repeatedOutcome.failure().isEmpty());
verify(ingestionFixture.hybridVectorService, never()).deleteByUrl(any(QdrantCollectionKind.class), anyString());
verify(ingestionFixture.ingestedFilePruneService, never())
.pruneObsoleteLocalStateAfterReplacement(anyString(), any(), any());
verify(ingestionFixture.fileIngestionMarkerStore, never())
.markFileIngested(anyString(), any(FileIngestionRecord.class));
verify(ingestionFixture.quarantineService, never()).quarantine(any(Path.class));
verify(ingestionFixture.chunkProcessingService, never())
.processAndStoreChunks(anyString(), anyString(), anyString(), anyString());
}

@Test
void shouldStopBeforeLaterFileWhenChunkStorageFails(@TempDir Path temporaryDirectory) throws IOException {
DocumentationSource documentationSource =
Expand Down