Hand out the checkpoint when the source is what died

.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

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

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<IngestBatch>(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)
 }
 

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<String>(), any<Sequence<PointStruct>>(), any()) } coAnswers {
+            if (firstBatch) {
+                firstBatch = false
+                kotlinx.coroutines.delay(50)
+            }
+        }
+        val seen = mutableListOf<IngestCheckpoint>()
+
+        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()