From 17f3c667c20b525b926a7013638b6dc62367babe Mon Sep 17 00:00:00 2001 From: TonyTonyCoder11 Date: Wed, 12 Aug 2026 00:14:36 +0200 Subject: [PATCH 1/2] Hand out the checkpoint when the source is what died An ingest whose source threw left the batches that were still in flight to be cancelled where they stood, because the exception propagated straight out of the coroutineScope. Whether the run reported any checkpoint at all then came down to which request happened to come back first, and a source that died early enough could leave the caller with no token whatsoever: the case a resume token exists for was the one case it was missing. The source's failure is now recorded and thrown after the workers have drained, 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, and re-sending an acknowledged point is free while skipping an unacknowledged one is silent data loss. This is what failed "an ingest killed partway resumes from its token" on the JVM REST run in CI: the assertion that a checkpoint was handed out at all. The linuxX64 run of the same suite failed in the same build with its case names truncated out of the console log, and the gRPC run of the same case passed, which is the shape a timing race has. The new unit test holds a batch in flight past the source's death, so the old behaviour fails it every time rather than now and then on a loaded runner. --- CHANGELOG.md | 10 +++++++ .../commonMain/kotlin/dev/kdrant/Ingest.kt | 19 +++++++++++++- .../jvmTest/kotlin/dev/kdrant/IngestTest.kt | 26 +++++++++++++++++++ 3 files changed, 54 insertions(+), 1 deletion(-) 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() From 9ce8ab17ac575a98ba5935fc4e38195d5b41433a Mon Sep 17 00:00:00 2001 From: TonyTonyCoder11 Date: Wed, 12 Aug 2026 00:14:36 +0200 Subject: [PATCH 2/2] Give the macOS runner three tries at downloading Qdrant The hosted runner intermittently fails the TLS handshake against the release host with "self signed certificate", on a certificate that verifies from everywhere else and verifies from the runner on the next attempt. There was no retry, so a hiccup lasting one second red-lined the whole job in nine. Three attempts with a growing pause, and the failure message names what could not be downloaded. Verification stays on: -k would trade a flake for a job that runs whatever answers the name. --- .github/workflows/ci.yml | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) 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