diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 827845b..cd136fa 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -159,8 +159,23 @@ jobs: # test. - name: Start Qdrant run: | - curl -fsSL -o qdrant.tar.gz \ - "https://github.com/qdrant/qdrant/releases/download/v${QDRANT_VERSION}/qdrant-aarch64-apple-darwin.tar.gz" + # The download is retried because the hosted runner's TLS to the release host fails now and + # then — "self signed certificate", on a certificate that is fine from everywhere else, and + # gone on the next attempt. Retrying is the honest answer to that; -k would turn a flake into + # a job that downloads and runs whatever answers the name. + downloaded="" + for attempt in 1 2 3; do + if curl -fsSL -o qdrant.tar.gz \ + "https://github.com/qdrant/qdrant/releases/download/v${QDRANT_VERSION}/qdrant-aarch64-apple-darwin.tar.gz" + then + downloaded=yes + break + fi + sleep $((attempt * 5)) + done + if [ -z "$downloaded" ]; then + echo "::error::could not download Qdrant ${QDRANT_VERSION} after three attempts"; exit 1 + fi tar -xzf qdrant.tar.gz ./qdrant & for _ in $(seq 1 60); do diff --git a/CHANGELOG.md b/CHANGELOG.md index f098473..bb438c6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,16 @@ All notable changes to this project are documented in this file. The format is b ## [Unreleased] +### Fixed + +- **An ingest whose source dies now hands out the checkpoint it earned.** The batches still in flight + when the source threw were cancelled where they stood, so whether a run reported any checkpoint at + all depended on which request happened to come back first, and a run killed early enough could report + none — leaving nothing to resume from in exactly the case the token exists for. The source's failure + is now held until those batches have drained and reported, then thrown, which is what a batch failure + already did and for the same reason: a batch cancelled after the server accepted it is a point the + collection holds and no token counts. + ## [2.2.0] - 2026-08-06 Tiers 8 and 9, complete. The theme is the distance between a request this client can build and one that diff --git a/kdrant-core/src/commonMain/kotlin/dev/kdrant/Ingest.kt b/kdrant-core/src/commonMain/kotlin/dev/kdrant/Ingest.kt index da2f935..fff3ab2 100644 --- a/kdrant-core/src/commonMain/kotlin/dev/kdrant/Ingest.kt +++ b/kdrant-core/src/commonMain/kotlin/dev/kdrant/Ingest.kt @@ -116,6 +116,9 @@ public data class IngestReport( * @throws IllegalArgumentException if a bound is not positive. * @throws KdrantException if a batch fails and cannot be retried. The exception is thrown after the * last [onCheckpoint] call, so the token in hand is the prefix that was written. + * @throws Throwable whatever [points] itself threw, on the same terms: the batches already in flight + * are allowed to finish and report before it reaches the caller, because a source that dies is the + * case a resume token exists for and it would be handed out empty otherwise. */ @Suppress("LongParameterList") public suspend fun QdrantClient.ingest( @@ -143,6 +146,14 @@ public suspend fun QdrantClient.ingest( // flight rather than by the size of the source. val queue = Channel(capacity = 0) + // The source's own failure is recorded here rather than thrown where it happens, for the reason a + // batch failure is: leaving the scope with an exception cancels the batches still in flight, and a + // batch cancelled after the server accepted it is a point the collection holds and no checkpoint + // counts. Worse, when the source dies early it can be the *only* batch, and the run that was killed + // at point four hundred thousand is handed no token at all. It is thrown once the workers have + // drained, so the last `onCheckpoint` has already run when the caller sees it. + var sourceFailure: Throwable? = null + coroutineScope { repeat(concurrency) { launch { @@ -168,12 +179,18 @@ public suspend fun QdrantClient.ingest( bytes += size } if (buffer.isNotEmpty()) queue.send(IngestBatch(index, offset, buffer)) + } catch (e: CancellationException) { + throw e + } catch (e: Throwable) { + sourceFailure = e } finally { queue.close() } } - tracker.failure?.let { throw it } + // The source first: when it is what died, a batch that failed afterwards is a consequence, and the + // caller is better told which of their two moving parts stopped the run. + (sourceFailure ?: tracker.failure)?.let { throw it } return IngestReport(tracker.checkpoint(), tracker.batchesSent) } diff --git a/kdrant-core/src/jvmTest/kotlin/dev/kdrant/IngestTest.kt b/kdrant-core/src/jvmTest/kotlin/dev/kdrant/IngestTest.kt index cce5dff..6fc37d2 100644 --- a/kdrant-core/src/jvmTest/kotlin/dev/kdrant/IngestTest.kt +++ b/kdrant-core/src/jvmTest/kotlin/dev/kdrant/IngestTest.kt @@ -7,6 +7,7 @@ import io.mockk.coEvery import io.mockk.mockk import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.asFlow +import kotlinx.coroutines.flow.flow import kotlinx.coroutines.test.runTest import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.Assertions.assertThrows @@ -137,6 +138,31 @@ class IngestTest { assertEquals(listOf(2L), seen.map { it.acknowledgedPoints }) } + @Test + fun `a source that dies mid-stream still hands out the batches that were in flight`() = runTest { + // The first batch is still on the wire when the source dies. Abandoning it there loses the only + // token the run ever had, and a process killed at point four hundred thousand starts from zero. + var firstBatch = true + coEvery { client.upsert(any(), any>(), any()) } coAnswers { + if (firstBatch) { + firstBatch = false + kotlinx.coroutines.delay(50) + } + } + val seen = mutableListOf() + + val source = flow { + points(1L..8L).collect { emit(it) } + error("the source died at point 9") + } + val failure = runCatching { + client.ingest("docs", source, batchSize = 2, concurrency = 2, onCheckpoint = { seen.add(it) }) + }.exceptionOrNull() + + assertTrue(failure is IllegalStateException, "the source's own failure is what the caller is told") + assertEquals(listOf(6L), seen.map { it.acknowledgedPoints }, "the in-flight batch never reported") + } + @Test fun `a resumed run skips what the token says was written`() = runTest { val batches = captureBatches()