fix(core): hold the last chunk until the blob and manifest are durable

Guarantee 1 is "durably stored in the configured bucket before the
response completes". The tee sent every chunk straight through, so the
client saw the last byte before the multipart upload was completed and
well before the manifest referenced it. A client that read a file and
immediately failed over to a sealed instance could be told the file had
never been seen -- and if either storage phase failed, it had already
received a clean 200 over bytes nothing recorded.

The tee now runs one chunk behind and releases the final chunk only
after the upload completes and the manifest write lands. Content-Length
is already on the wire, so a client cannot treat the transfer as
complete without those bytes; a storage or manifest failure now ends the
body short instead, which is what the client detects and retries on.

Costs one chunk of extra latency at the end of a transfer and nothing
else: the withheld chunk is bounded by the upstream chunk size, and a
disconnected client still has its transfer finished without it.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XZG2i4AmfSqE97EJGBVb64

crates/rustingface-api/tests/sovereignty.rs

@@ -143,6 +143,44 @@ async fn a_cold_fetch_serves_the_bytes_and_leaves_them_in_the_bucket() {
     assert_eq!(std::fs::read(&blob).unwrap(), weights);
 }
 
+#[tokio::test]
+async fn the_response_does_not_complete_until_the_bytes_are_recorded() {
+    let hub = FakeHub::start().await;
+    let weights = publish_model(&hub).await;
+    hub.throttle(8 * 1024, Duration::from_millis(2)).await;
+
+    let bucket = bucket_dir("durable-before-complete");
+    let rf = instance(&bucket, Some(&hub), |_| {});
+
+    // Guarantee 1 is "durably stored before the response completes", not
+    // "shortly after". The moment the client has read the last byte, a second
+    // instance with no upstream at all must already be able to serve it.
+    let (status, _, body) = call(
+        &rf.app,
+        "GET",
+        "/Qwen/Qwen3-32B/resolve/main/model.safetensors",
+        &[],
+    )
+    .await;
+    assert_eq!(status, StatusCode::OK);
+    assert_eq!(body, weights);
+
+    let sealed = instance(&bucket, None, |_| {});
+    let (status, _, replayed) = call(
+        &sealed.app,
+        "GET",
+        "/Qwen/Qwen3-32B/resolve/main/model.safetensors",
+        &[],
+    )
+    .await;
+    assert_eq!(
+        status,
+        StatusCode::OK,
+        "the file was served but is not yet servable offline"
+    );
+    assert_eq!(replayed, weights);
+}
+
 #[tokio::test]
 async fn a_sealed_instance_serves_the_same_bytes_with_no_upstream_at_all() {
     let hub = FakeHub::start().await;

crates/rustingface-core/src/fetch.rs

@@ -150,6 +150,14 @@ impl Transfer {
         let mut hasher = Digest::new();
         let mut client = Some(client);
         let mut buffered: Vec<Bytes> = Vec::new();
+        // The last chunk is withheld from the client until the blob is durable
+        // and the manifest references it. Content-Length is already on the
+        // wire, so a client cannot consider the transfer complete without
+        // these bytes -- which is what makes "durably stored before the
+        // response completes" true rather than nearly true. If storage fails,
+        // the client gets a body that ends short instead of a clean 200 over
+        // bytes nothing recorded.
+        let mut holdback: Option<Bytes> = None;
 
         let mut writer = match &self.sink {
             Sink::Buffered => None,
@@ -168,10 +176,12 @@ impl Transfer {
                     None => buffered.push(chunk.clone()),
                 }
 
-                // Tee to the client, dropping it — not the transfer — if it has
-                // gone away or stopped reading.
-                if let Some(tx) = &client {
-                    match tokio::time::timeout(self.stall_timeout, tx.send(Ok(chunk))).await {
+                // Tee to the client one chunk behind, dropping the client --
+                // not the transfer -- if it has gone away or stopped reading.
+                if let Some(previous) = holdback.replace(chunk)
+                    && let Some(tx) = &client
+                {
+                    match tokio::time::timeout(self.stall_timeout, tx.send(Ok(previous))).await {
                         Ok(Ok(())) => {}
                         Ok(Err(_)) => {
                             debug!(path = %self.path, "client disconnected; finishing the transfer without it");
@@ -228,7 +238,14 @@ impl Transfer {
         // Phase 1: the blob is durable. A crash after this point leaks an
         // orphan, which `gc` reclaims; nothing references it yet.
         match (writer, &self.sink) {
-            (Some(writer), Sink::Direct { .. }) => writer.finish().await?,
+            (Some(writer), Sink::Direct { .. }) => {
+                if let Err(err) = writer.finish().await {
+                    if let Some(tx) = &client {
+                        let _ = tx.send(Err(Error::Storage(err.to_string()))).await;
+                    }
+                    return Err(err);
+                }
+            }
             (Some(writer), Sink::Staged { key: staged }) => {
                 writer.finish().await?;
                 if self.store.head(&final_key).await?.is_none() {
@@ -264,14 +281,31 @@ impl Transfer {
 
         // Phase 2: only now may a manifest reference it.
         let entry = entry_from(&self.path, size, &hex, &self.head);
-        registry
+        if let Err(err) = registry
             .record_entry(
                 self.registry_repo_type,
                 &self.repo_id,
                 &self.commit,
                 entry.clone(),
             )
-            .await?;
+            .await
+        {
+            // The blob is stored but unreferenced -- an orphan `gc` reclaims.
+            // The client must not be told the transfer succeeded, because a
+            // later sealed read of this path would 404.
+            if let Some(tx) = &client {
+                let _ = tx.send(Err(Error::Storage(err.to_string()))).await;
+            }
+            return Err(err);
+        }
+
+        // Everything is durable and referenced; only now may the client be
+        // told the transfer is complete.
+        if let Some(last) = holdback.take()
+            && let Some(tx) = &client
+        {
+            let _ = tokio::time::timeout(self.stall_timeout, tx.send(Ok(last))).await;
+        }
 
         info!(
             repo_id = %self.repo_id, commit = %self.commit, path = %self.path, size,