S3: commit GET status only after the first read succeeds (#10930)

streamFromVolumeServers wrote the 200/206 status from filer metadata
before any byte had been fetched from a volume server, so a missing or
corrupted needle surfaced as a broken 200 body and the request metrics
recorded a success. Defer the status commit to the first body write: a
failed first read now returns a clean 500 before headers, while the
wire timing of successful responses is unchanged since net/http buffers
the status line until body bytes arrive anyway.

weed/s3api/s3api_local_read_fallback_test.go

@@ -166,7 +166,7 @@ func TestS3CachedReadFallsBackToRemote(t *testing.T) {
 		assert.Equal(t, int64(4), client.gotSize)
 	})
 
-	t.Run("a local-only object keeps the existing error", func(t *testing.T) {
+	t.Run("a local-only object gets a clean 500, not a broken 200 body", func(t *testing.T) {
 		client := &fakeStreamRemoteClient{data: content}
 		remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: client}
 		s3a := newLocalReadFallbackServer(t, startStreamThroughFiler(t, "faketest-nofallback", nil))
@@ -180,6 +180,8 @@ func TestS3CachedReadFallsBackToRemote(t *testing.T) {
 		err := s3a.streamFromVolumeServers(w, r, local, "", "mybucket", "dir/obj.bin", "")
 
 		require.Error(t, err)
+		assert.Equal(t, http.StatusInternalServerError, w.Code, "the status must not be committed before the first read succeeds")
+		assert.Contains(t, w.Body.String(), "InternalError")
 		assert.Nil(t, client.gotLoc, "an object that is not remote-mounted has no remote to fall back to")
 	})
 
@@ -201,6 +203,52 @@ func TestS3CachedReadFallsBackToRemote(t *testing.T) {
 	})
 }
 
+// the deferred status commit must be invisible on the success path: a readable
+// local object still streams with the same status and headers as before
+func TestS3LocalReadCommitsStatusOnFirstWrite(t *testing.T) {
+	content := []byte("0123456789")
+	newLocalServer := func(t *testing.T, name string) *S3ApiServer {
+		s3a := newLocalReadFallbackServer(t, startStreamThroughFiler(t, name, nil))
+		cache := chunk_cache.NewChunkCacheInMemory(16)
+		cache.SetChunk("1,0123456789ab", content)
+		s3a.readerCache = filer.NewReaderCache(8, cache, s3a.filerClient.GetLookupFileIdFunction(), s3a.filerClient)
+		return s3a
+	}
+	localEntry := func() *filer_pb.Entry {
+		local := cachedEntry(content)
+		local.RemoteEntry = nil
+		return local
+	}
+
+	t.Run("readable local object streams a 200", func(t *testing.T) {
+		s3a := newLocalServer(t, "faketest-localok")
+		w := httptest.NewRecorder()
+		r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil)
+
+		err := s3a.streamFromVolumeServers(w, r, localEntry(), "", "mybucket", "dir/obj.bin", "")
+
+		require.NoError(t, err)
+		assert.Equal(t, http.StatusOK, w.Code)
+		assert.Equal(t, content, w.Body.Bytes())
+		assert.Equal(t, "10", w.Header().Get("Content-Length"))
+	})
+
+	t.Run("readable range streams a 206 with range headers", func(t *testing.T) {
+		s3a := newLocalServer(t, "faketest-localrange")
+		w := httptest.NewRecorder()
+		r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil)
+		r.Header.Set("Range", "bytes=2-5")
+
+		err := s3a.streamFromVolumeServers(w, r, localEntry(), "", "mybucket", "dir/obj.bin", "")
+
+		require.NoError(t, err)
+		assert.Equal(t, http.StatusPartialContent, w.Code)
+		assert.Equal(t, content[2:6], w.Body.Bytes())
+		assert.Equal(t, "bytes 2-5/10", w.Header().Get("Content-Range"))
+		assert.Equal(t, "4", w.Header().Get("Content-Length"))
+	})
+}
+
 // a multi-chunk object whose later chunk's volume server is unreachable
 type truncatedReaderAt struct {
 	data    []byte

weed/s3api/s3api_object_handlers.go

@@ -57,6 +57,23 @@ func (cw *countingWriter) Write(p []byte) (int, error) {
 	return n, err
 }
 
+// commitOnFirstWrite defers the status line until the first body write, so the
+// first read from the volume servers must succeed before the 200 is committed
+// and a missing needle still gets a clean 5xx instead of a broken 200 body.
+type commitOnFirstWrite struct {
+	w         io.Writer
+	commit    func()
+	committed bool
+}
+
+func (c *commitOnFirstWrite) Write(p []byte) (int, error) {
+	if !c.committed {
+		c.committed = true
+		c.commit()
+	}
+	return c.w.Write(p)
+}
+
 // adjustRangeForPart adjusts a client's Range header to absolute offsets within a part.
 // Parameters:
 //   - partStartOffset: the absolute start offset of the part in the object
@@ -1155,41 +1172,48 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
 		}
 	}
 
-	// All validation and preparation successful - NOW set headers and write status
-	tHeaderSet := time.Now()
-	s3a.setResponseHeaders(w, r, entry, totalSize)
-
-	// Override/add range-specific headers if this is a range request
-	if isRangeRequest {
-		w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, offset+size-1, totalSize))
+	// Headers and status are committed on the first body write, once the first
+	// read from the volume servers has succeeded. The client sees the same wire
+	// timing either way -- the status line sits in the server's write buffer
+	// until body bytes arrive -- but a first read that fails now surfaces as a
+	// clean 5xx instead of a 200 with a broken body.
+	body := &commitOnFirstWrite{w: w, commit: func() {
+		tHeaderSet := time.Now()
+		s3a.setResponseHeaders(w, r, entry, totalSize)
+		if isRangeRequest {
+			w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, offset+size-1, totalSize))
+		}
 		w.Header().Set("Content-Length", strconv.FormatInt(size, 10))
-	} else {
-		w.Header().Set("Content-Length", strconv.FormatInt(size, 10))
-	}
-	headerSetTime = time.Since(tHeaderSet)
-
-	// Now write status code (headers are all set, stream is ready)
-	if isRangeRequest {
-		w.WriteHeader(http.StatusPartialContent)
-	} else {
-		w.WriteHeader(http.StatusOK)
-	}
-
-	// Track time to first byte metric
-	TimeToFirstByte(r.Method, t0, r)
+		headerSetTime = time.Since(tHeaderSet)
+		if isRangeRequest {
+			w.WriteHeader(http.StatusPartialContent)
+		} else {
+			w.WriteHeader(http.StatusOK)
+		}
+		TimeToFirstByte(r.Method, t0, r)
+	}}
 
 	// Stream directly to response with counting wrapper.
 	// ChunkReadAt's ReadAt is backed by in-memory prefetched chunk buffers, so
 	// io.CopyBuffer drains them as fast memcpys.
 	tStreamExec := time.Now()
 	glog.V(4).Infof("streamFromVolumeServers: starting chunk reader, offset=%d, size=%d", offset, size)
-	written, err := s3a.streamRangeToClient(w, r, reader, entry, bucket, object, offset, size, totalSize, versionId)
+	written, err := s3a.streamRangeToClient(body, r, reader, entry, bucket, object, offset, size, totalSize, versionId)
 	streamExecTime = time.Since(tStreamExec)
 	// Track traffic even on partial writes for accurate egress accounting
 	if written > 0 {
 		BucketTrafficSent(written, r)
 	}
 	if err != nil {
+		if !body.committed {
+			if isCanceledStreamingError(err) {
+				glog.V(3).Infof("streamFromVolumeServers: request canceled before first byte of %s/%s: %v", bucket, object, err)
+				return err
+			}
+			glog.Errorf("streamFromVolumeServers: first read of %s/%s failed: %v", bucket, object, err)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+			return newStreamErrorWithResponse(err)
+		}
 		switch {
 		case isCanceledStreamingError(err):
 			// Client disconnected mid-stream (e.g. Nginx upstream timeout, browser cancel) - expected
@@ -1203,6 +1227,11 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
 		// Streaming error after WriteHeader was called - response already partially written
 		return newStreamErrorWithResponse(err)
 	}
+	if !body.committed {
+		// a zero-length body never writes, so commit the empty response here
+		body.committed = true
+		body.commit()
+	}
 	glog.V(4).Infof("streamFromVolumeServers: streamFn completed successfully, wrote %d bytes", written)
 	return nil
 }