From d9d5fab35b819db37dfec007d217dd278f628e7c Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 24 Aug 2026 16:14:52 -0700 Subject: [PATCH] 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 | 50 +++++++++++++- weed/s3api/s3api_object_handlers.go | 73 ++++++++++++++------ 2 files changed, 100 insertions(+), 23 deletions(-) diff --git a/weed/s3api/s3api_local_read_fallback_test.go b/weed/s3api/s3api_local_read_fallback_test.go index c692ff336..6294ca1c0 100644 --- a/weed/s3api/s3api_local_read_fallback_test.go +++ b/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 diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index 43c5d75db..43544043b 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/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 }