S3: optionally serve remote-mounted objects from remote when the local read fails (#10837)

* feat(s3): serve from remote on local read failure

When a locally-cached chunk of a remote-mounted object becomes unreadable
(volume server down/restarting, or an evicted needle 404ing under
retry-backoff), fall back to serving the object from its mounted remote
instead of erroring. A bounded pre-flight probe makes a stuck volume trip
the timeout rather than stalling the request.

Gated by -localReadFallbackToRemote (default off) with
-localReadFallbackTimeout (2s default), so existing deployments are
unaffected until they opt in.

* fix(s3): register local-read-fallback flags for mini/server/filer

The mini, server and filer launchers build S3Options directly and only
populate the flag pointers they register. Without registering the two new
flags there, startS3Server dereferenced nil pointers and crashed at boot,
failing every integration suite that runs `weed mini`.

* fix(s3): treat a zero-byte probe read as unreadable

A read that returns no byte -- whether it reports io.EOF or no error at all
-- means the offset is not locally readable, so the probe must fall back to
the remote rather than proceeding to stream a truncated response. Only a
returned byte (including the object's final byte with a trailing io.EOF)
counts as readable.

* s3: finish a mid-stream local read failure from the remote mount

The pre-flight probe only proves the byte at the requested offset readable.
A multi-chunk object can still lose a later chunk after the 200/206 and its
Content-Length are committed, which truncated the body with no fallback.
Resume from the mounted remote at the byte the local copy stopped at, so the
response still carries the declared length. A short local read that surfaces
as a clean EOF is treated the same way instead of silently truncating.

* s3: fall back to the remote mount without a CLI switch

Serving a remote-mounted object from its authoritative remote is what the
read should have done all along -- the alternative is a 500 on an object the
cluster can still reach -- so make it the behavior instead of two new flags,
with the probe bounded by a constant.

* s3: trim the comments on the fallback path

* filer: report only the contiguous prefix when a parallel chunk read fails

The parallel branch of doReadAt fans the chunk reads straight into their own
windows of the output buffer, then sums every task's bytesRead. A middle chunk
failing while a later one succeeds therefore returned a length covering a hole
the reader never filled, handing the caller zeros in the middle of otherwise
valid data.

* s3: only splice the remote onto a local prefix while it is the cached generation

Eligibility establishes a size match, not byte identity: a remote key
overwritten with same-size content between the cache fill and the fallback
would have finished the response with bytes from a second generation, under
the first one's ETag. Stat the remote before resuming and keep the local
error when it no longer matches -- a truncated body is a visible failure,
a spliced one is not.

---------

Co-authored-by: Chris Lu <chrislusf@users.noreply.github.com>
Co-authored-by: Chris Lu <chris.lu@gmail.com>

weed/filer/reader_at.go

@@ -288,6 +288,15 @@ func (c *ChunkReadAt) doReadAt(ctx context.Context, p []byte, offset int64) (n i
 		}
 
 		if err != nil {
+			// a failed chunk leaves its window untouched while the tasks after it
+			// may well have filled theirs, so only the prefix up to the hole is
+			// data the caller may use
+			for _, task := range tasks {
+				if int64(task.bytesRead) != task.bufferEnd-task.bufferStart {
+					n = int(task.bufferStart) + task.bytesRead
+					break
+				}
+			}
 			return n, ts, err
 		}
 	}

weed/filer/reader_at_test.go

@@ -3,10 +3,14 @@ package filer
 import (
 	"bytes"
 	"context"
+	"errors"
 	"io"
 	"math"
 	"strconv"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 type mockChunkCache struct {
@@ -202,3 +206,54 @@ func TestReaderAtSparseFileDoesNotLeak(t *testing.T) {
 	testReadAt(t, readerAt, 0, 3, 3, io.EOF, []byte{2, 2, 2}, []byte{0, 0, 0})
 	testReadAt(t, readerAt, 1, 2, 2, io.EOF, []byte{2, 2}, []byte{0, 0})
 }
+
+// holeChunkCache serves every chunk but one, so that chunk falls through to a
+// lookup that fails.
+type holeChunkCache struct {
+	mockChunkCache
+	missingFileId string
+}
+
+func (c *holeChunkCache) ReadChunkAt(data []byte, fileId string, offset uint64) (int, error) {
+	if fileId == c.missingFileId {
+		return 0, nil
+	}
+	return c.mockChunkCache.ReadChunkAt(data, fileId, offset)
+}
+
+func (c *holeChunkCache) GetMaxFilePartSizeInCache() uint64 {
+	return math.MaxUint64
+}
+
+// The parallel reads place their bytes directly in the output buffer, so a
+// failing middle chunk leaves a hole with valid data after it. Reporting that
+// length would hand the caller bytes it never read.
+func TestReaderAtParallelFailureReturnsContiguousPrefix(t *testing.T) {
+	visibles := NewIntervalList[*VisibleInterval]()
+	for i, fileId := range []string{"1", "2", "3"} {
+		addVisibleInterval(visibles, &VisibleInterval{
+			start:     int64(i) * 4,
+			stop:      int64(i)*4 + 4,
+			fileId:    fileId,
+			chunkSize: 4,
+		})
+	}
+
+	readerAt := &ChunkReadAt{
+		ctx:        context.Background(),
+		chunkViews: ViewFromVisibleIntervals(visibles, 0, math.MaxInt64),
+		fileSize:   12,
+		readerCache: NewReaderCache(3, &holeChunkCache{missingFileId: "2"}, func(ctx context.Context, fileId string) ([]string, error) {
+			return nil, errors.New("volume down")
+		}, nil),
+		readerPattern: NewReaderPattern(),
+		prefetchCount: 3,
+	}
+
+	buf := make([]byte, 12)
+	n, err := readerAt.ReadAt(buf, 0)
+
+	require.Error(t, err)
+	assert.Equal(t, 4, n, "only the bytes before the failed chunk are readable")
+	assert.Equal(t, []byte{1, 1, 1, 1}, buf[:n])
+}

weed/s3api/s3api_local_read_fallback_test.go

@@ -0,0 +1,318 @@
+package s3api
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/remote_storage"
+	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
+	"github.com/seaweedfs/seaweedfs/weed/util/chunk_cache"
+	"github.com/seaweedfs/seaweedfs/weed/wdclient"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// stubReaderAt stands in for ChunkReadAt in probe unit tests.
+type stubReaderAt struct {
+	n   int
+	err error
+}
+
+func (s stubReaderAt) ReadAtWithTime(ctx context.Context, p []byte, offset int64) (int, int64, error) {
+	return s.n, 0, s.err
+}
+
+// blockingReaderAt never returns until the context is cancelled.
+type blockingReaderAt struct{}
+
+func (blockingReaderAt) ReadAtWithTime(ctx context.Context, p []byte, offset int64) (int, int64, error) {
+	<-ctx.Done()
+	return 0, 0, ctx.Err()
+}
+
+func TestProbeReadable(t *testing.T) {
+	t.Run("readable local copy", func(t *testing.T) {
+		assert.NoError(t, probeReadable(context.Background(), stubReaderAt{n: 1}, 0, time.Second))
+	})
+
+	t.Run("final byte with trailing EOF counts as readable", func(t *testing.T) {
+		assert.NoError(t, probeReadable(context.Background(), stubReaderAt{n: 1, err: io.EOF}, 0, time.Second))
+	})
+
+	t.Run("zero-byte EOF is unreadable", func(t *testing.T) {
+		assert.Error(t, probeReadable(context.Background(), stubReaderAt{n: 0, err: io.EOF}, 0, time.Second))
+	})
+
+	t.Run("zero-byte read without error is unreadable", func(t *testing.T) {
+		assert.ErrorIs(t, probeReadable(context.Background(), stubReaderAt{n: 0, err: nil}, 0, time.Second), io.ErrUnexpectedEOF)
+	})
+
+	t.Run("read error surfaces", func(t *testing.T) {
+		boom := errors.New("volume: connection refused")
+		assert.ErrorIs(t, probeReadable(context.Background(), stubReaderAt{err: boom}, 0, time.Second), boom)
+	})
+
+	t.Run("stuck volume trips the timeout instead of blocking", func(t *testing.T) {
+		start := time.Now()
+		err := probeReadable(context.Background(), blockingReaderAt{}, 0, 50*time.Millisecond)
+		assert.ErrorIs(t, err, context.DeadlineExceeded)
+		assert.Less(t, time.Since(start), time.Second, "must not block past the probe timeout")
+	})
+}
+
+func TestShouldFallBackToRemote(t *testing.T) {
+	const size = int64(100)
+	remoteEntry := func() *filer_pb.Entry {
+		return &filer_pb.Entry{
+			Attributes:  &filer_pb.FuseAttributes{FileSize: uint64(size)},
+			RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: size},
+		}
+	}
+	server := func() *S3ApiServer {
+		return &S3ApiServer{option: &S3ApiServerOption{}}
+	}
+
+	t.Run("matching remote", func(t *testing.T) {
+		assert.True(t, server().shouldFallBackToRemote(remoteEntry(), size, ""))
+	})
+
+	t.Run("null versionId is treated as unversioned", func(t *testing.T) {
+		assert.True(t, server().shouldFallBackToRemote(remoteEntry(), size, "null"))
+	})
+
+	t.Run("versioned read cannot use the unversioned remote key", func(t *testing.T) {
+		assert.False(t, server().shouldFallBackToRemote(remoteEntry(), size, "v123"))
+	})
+
+	t.Run("latest read that resolves to a version cannot use the unversioned key", func(t *testing.T) {
+		versioned := remoteEntry()
+		versioned.Extended = map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v9")}
+		assert.False(t, server().shouldFallBackToRemote(versioned, size, ""))
+	})
+
+	t.Run("no remote entry", func(t *testing.T) {
+		local := &filer_pb.Entry{Attributes: &filer_pb.FuseAttributes{FileSize: uint64(size)}}
+		assert.False(t, server().shouldFallBackToRemote(local, size, ""))
+	})
+
+	t.Run("size mismatch is not served as identical bytes", func(t *testing.T) {
+		assert.False(t, server().shouldFallBackToRemote(remoteEntry(), size+1, ""))
+	})
+}
+
+// The fake filer does not implement LookupVolume, so every chunk read through
+// this ReaderCache fails, standing in for an unreachable volume server.
+func newLocalReadFallbackServer(t *testing.T, filerAddr pb.ServerAddress) *S3ApiServer {
+	t.Helper()
+	s3a := newRemoteCacheTestServer(filerAddr)
+	fc := wdclient.NewFilerClient(s3a.option.Filers, s3a.option.GrpcDialOption, s3a.option.DataCenter)
+	s3a.filerClient = fc
+	s3a.readerCache = filer.NewReaderCache(8, (*chunk_cache.TieredChunkCache)(nil), fc.GetLookupFileIdFunction(), fc)
+	return s3a
+}
+
+// a remote-mounted object whose cached copy is in the metadata but unreadable
+func cachedEntry(content []byte) *filer_pb.Entry {
+	return &filer_pb.Entry{
+		Name:        "obj.bin",
+		Attributes:  &filer_pb.FuseAttributes{FileSize: uint64(len(content))},
+		Chunks:      []*filer_pb.FileChunk{{FileId: "1,0123456789ab", Size: uint64(len(content)), Offset: 0}},
+		RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: int64(len(content))},
+	}
+}
+
+func TestS3CachedReadFallsBackToRemote(t *testing.T) {
+	content := []byte("0123456789")
+
+	t.Run("unreadable local copy is served from the mounted remote", func(t *testing.T) {
+		client := &fakeStreamRemoteClient{data: content}
+		remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: client}
+		s3a := newLocalReadFallbackServer(t, startStreamThroughFiler(t, "faketest-cachedfail", nil))
+		w := httptest.NewRecorder()
+		r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil)
+
+		err := s3a.streamFromVolumeServers(w, r, cachedEntry(content), "", "mybucket", "dir/obj.bin", "")
+
+		require.NoError(t, err)
+		assert.Equal(t, http.StatusOK, w.Code)
+		assert.Equal(t, content, w.Body.Bytes())
+		require.NotNil(t, client.gotLoc, "must read from the mounted remote")
+		assert.Equal(t, "/data/dir/obj.bin", client.gotLoc.Path)
+	})
+
+	t.Run("range fallback serves the requested window from the remote", func(t *testing.T) {
+		client := &fakeStreamRemoteClient{data: content}
+		remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: client}
+		s3a := newLocalReadFallbackServer(t, startStreamThroughFiler(t, "faketest-cachedrange", nil))
+		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, cachedEntry(content), "", "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, int64(2), client.gotOffset)
+		assert.Equal(t, int64(4), client.gotSize)
+	})
+
+	t.Run("a local-only object keeps the existing error", func(t *testing.T) {
+		client := &fakeStreamRemoteClient{data: content}
+		remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: client}
+		s3a := newLocalReadFallbackServer(t, startStreamThroughFiler(t, "faketest-nofallback", nil))
+		local := cachedEntry(content)
+		local.RemoteEntry = nil
+		w := httptest.NewRecorder()
+		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+		defer cancel()
+		r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil).WithContext(ctx)
+
+		err := s3a.streamFromVolumeServers(w, r, local, "", "mybucket", "dir/obj.bin", "")
+
+		require.Error(t, err)
+		assert.Nil(t, client.gotLoc, "an object that is not remote-mounted has no remote to fall back to")
+	})
+
+	t.Run("versioned read does not fall back to the unversioned remote key", func(t *testing.T) {
+		client := &fakeStreamRemoteClient{data: content}
+		remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: client}
+		s3a := newLocalReadFallbackServer(t, startStreamThroughFiler(t, "faketest-cachedversioned", nil))
+		versioned := cachedEntry(content)
+		versioned.Extended = map[string][]byte{s3_constants.ExtVersionIdKey: []byte("v456")}
+		w := httptest.NewRecorder()
+		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+		defer cancel()
+		r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin?versionId=v456", nil).WithContext(ctx)
+
+		err := s3a.streamFromVolumeServers(w, r, versioned, "", "mybucket", "dir/obj.bin", "v456")
+
+		require.Error(t, err)
+		assert.Nil(t, client.gotLoc, "a versioned read must not serve the unversioned remote object")
+	})
+}
+
+// a multi-chunk object whose later chunk's volume server is unreachable
+type truncatedReaderAt struct {
+	data    []byte
+	breakAt int64
+	err     error
+}
+
+func (t truncatedReaderAt) ReadAt(p []byte, offset int64) (int, error) {
+	if offset >= t.breakAt {
+		return 0, t.err
+	}
+	return copy(p, t.data[offset:t.breakAt]), nil
+}
+
+// a client that disconnected mid-response
+type failingWriter struct{ err error }
+
+func (f failingWriter) Write(p []byte) (int, error) { return 0, f.err }
+
+// the window the probe cannot cover: the byte at offset reads fine, the response
+// is committed, and only then does a later chunk turn out to be unreadable
+func TestStreamRangeToClientFinishesFromRemote(t *testing.T) {
+	content := []byte("0123456789")
+	size := int64(len(content))
+	newServer := func(t *testing.T, name string) (*S3ApiServer, *fakeStreamRemoteClient) {
+		client := &fakeStreamRemoteClient{data: content}
+		remote_storage.RemoteStorageClientMakers["faketest"] = &fakeStreamRemoteMaker{client: client}
+		return newLocalReadFallbackServer(t, startStreamThroughFiler(t, name, nil)), client
+	}
+
+	t.Run("later chunk failure is finished from the remote", func(t *testing.T) {
+		s3a, client := newServer(t, "faketest-midstream")
+		w := httptest.NewRecorder()
+		r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil)
+		local := truncatedReaderAt{data: content, breakAt: 4, err: errors.New("volume: connection refused")}
+
+		written, err := s3a.streamRangeToClient(w, r, local, cachedEntry(content), "mybucket", "dir/obj.bin", 0, size, size, "")
+
+		require.NoError(t, err)
+		assert.Equal(t, size, written)
+		assert.Equal(t, content, w.Body.Bytes())
+		assert.Equal(t, int64(4), client.gotOffset, "the remote must resume where the local copy stopped")
+		assert.Equal(t, int64(6), client.gotSize)
+	})
+
+	t.Run("a short local read reported as clean EOF is not served truncated", func(t *testing.T) {
+		s3a, client := newServer(t, "faketest-shortread")
+		w := httptest.NewRecorder()
+		r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil)
+		local := truncatedReaderAt{data: content, breakAt: 7, err: io.EOF}
+
+		written, err := s3a.streamRangeToClient(w, r, local, cachedEntry(content), "mybucket", "dir/obj.bin", 0, size, size, "")
+
+		require.NoError(t, err)
+		assert.Equal(t, size, written)
+		assert.Equal(t, content, w.Body.Bytes())
+		assert.Equal(t, int64(7), client.gotOffset)
+	})
+
+	t.Run("range read resumes at the absolute offset", func(t *testing.T) {
+		s3a, client := newServer(t, "faketest-midstreamrange")
+		w := httptest.NewRecorder()
+		r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil)
+		local := truncatedReaderAt{data: content, breakAt: 5, err: errors.New("volume: connection refused")}
+
+		written, err := s3a.streamRangeToClient(w, r, local, cachedEntry(content), "mybucket", "dir/obj.bin", 2, 6, size, "")
+
+		require.NoError(t, err)
+		assert.Equal(t, int64(6), written)
+		assert.Equal(t, content[2:8], w.Body.Bytes())
+		assert.Equal(t, int64(5), client.gotOffset)
+		assert.Equal(t, int64(3), client.gotSize)
+	})
+
+	t.Run("a remote overwritten since caching is not spliced onto the local prefix", func(t *testing.T) {
+		s3a, client := newServer(t, "faketest-midstreamchanged")
+		client.stat = &filer_pb.RemoteEntry{RemoteSize: size, RemoteETag: "reuploaded"}
+		w := httptest.NewRecorder()
+		r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil)
+		boom := errors.New("volume: connection refused")
+
+		written, err := s3a.streamRangeToClient(w, r, truncatedReaderAt{data: content, breakAt: 4, err: boom}, cachedEntry(content), "mybucket", "dir/obj.bin", 0, size, size, "")
+
+		require.Error(t, err)
+		assert.Equal(t, int64(4), written, "a truncated body beats one mixing two generations")
+		assert.Nil(t, client.gotLoc, "must not read a remote that no longer matches what was cached")
+	})
+
+	t.Run("a local-only object keeps the mid-stream error", func(t *testing.T) {
+		s3a, client := newServer(t, "faketest-midstreamlocal")
+		local := cachedEntry(content)
+		local.RemoteEntry = nil
+		w := httptest.NewRecorder()
+		r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil)
+		boom := errors.New("volume: connection refused")
+
+		written, err := s3a.streamRangeToClient(w, r, truncatedReaderAt{data: content, breakAt: 4, err: boom}, local, "mybucket", "dir/obj.bin", 0, size, size, "")
+
+		assert.ErrorIs(t, err, boom)
+		assert.Equal(t, int64(4), written)
+		assert.Nil(t, client.gotLoc, "an object that is not remote-mounted has no remote to fall back to")
+	})
+
+	t.Run("a failed write to the client is not refetched from the remote", func(t *testing.T) {
+		s3a, client := newServer(t, "faketest-midstreamwrite")
+		r := httptest.NewRequest(http.MethodGet, "/mybucket/dir/obj.bin", nil)
+		broken := errors.New("write: broken pipe")
+
+		written, err := s3a.streamRangeToClient(failingWriter{err: broken}, r, bytes.NewReader(content), cachedEntry(content), "mybucket", "dir/obj.bin", 0, size, size, "")
+
+		assert.ErrorIs(t, err, broken)
+		assert.Zero(t, written)
+		assert.Nil(t, client.gotLoc, "the client is gone; refetching from the remote is wasted work")
+	})
+}

weed/s3api/s3api_object_handlers.go

@@ -43,13 +43,17 @@ var zeroBuf = make([]byte, 32*1024)
 
 // countingWriter wraps an io.Writer to count bytes written
 type countingWriter struct {
-	w       io.Writer
-	written int64
+	w        io.Writer
+	written  int64
+	writeErr error
 }
 
 func (cw *countingWriter) Write(p []byte) (int, error) {
 	n, err := cw.w.Write(p)
 	cw.written += int64(n)
+	if err != nil {
+		cw.writeErr = err
+	}
 	return n, err
 }
 
@@ -1037,33 +1041,11 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
 				// latest-version read -- has no origin key, so it keeps the error
 				// paths below.
 				if cacheVersionId == "" || cacheVersionId == "null" {
-					if remoteReader, remoteErr := s3a.openRemoteStream(r.Context(), bucket, object, offset, size); remoteErr == nil {
-						defer remoteReader.Close()
-						s3a.setResponseHeaders(w, r, entry, totalSize)
-						w.Header().Set("Content-Length", strconv.FormatInt(size, 10))
-						if isRangeRequest {
-							w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, offset+size-1, totalSize))
-							w.WriteHeader(http.StatusPartialContent)
-						} else {
-							w.WriteHeader(http.StatusOK)
-						}
-						TimeToFirstByte(r.Method, t0, r)
-						cw := &countingWriter{w: w}
-						_, copyErr := io.CopyN(cw, remoteReader, size)
-						if copyErr == io.EOF {
-							// the origin returned fewer bytes than the entry's RemoteSize
-							copyErr = io.ErrUnexpectedEOF
-						}
-						if cw.written > 0 {
-							BucketTrafficSent(cw.written, r)
-						}
-						if copyErr != nil {
-							glog.V(2).Infof("streamFromVolumeServers: origin stream %s/%s ended after %d bytes: %v", bucket, object, cw.written, copyErr)
-							return newStreamErrorWithResponse(copyErr)
+					if served, streamErr := s3a.serveObjectFromRemoteMount(w, r, entry, bucket, object, offset, size, isRangeRequest, totalSize, t0); served {
+						if streamErr != nil {
+							return newStreamErrorWithResponse(streamErr)
 						}
 						return nil
-					} else {
-						glog.Warningf("streamFromVolumeServers: origin stream %s/%s: %v", bucket, object, remoteErr)
 					}
 				}
 				// Origin unreadable. A permanent cache error is final; a transient one
@@ -1119,6 +1101,15 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
 		} else {
 			glog.Errorf("streamFromVolumeServers: failed to resolve chunks: %v", err)
 		}
+		if s3a.shouldFallBackToRemote(entry, totalSize, versionId) {
+			glog.V(1).Infof("streamFromVolumeServers: resolving chunks for %s/%s failed, falling back to remote mount", bucket, object)
+			if served, streamErr := s3a.serveObjectFromRemoteMount(w, r, entry, bucket, object, offset, size, isRangeRequest, totalSize, t0); served {
+				if streamErr != nil {
+					return newStreamErrorWithResponse(streamErr)
+				}
+				return nil
+			}
+		}
 		// Write S3-compliant XML error response
 		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return newStreamErrorWithResponse(fmt.Errorf("failed to resolve chunks: %v", err))
@@ -1142,6 +1133,28 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
 	reader := filer.NewChunkReaderAtFromClient(ctx, s3a.readerCache, chunkViews, totalSize, filer.DefaultPrefetchCount)
 	streamPrepTime = time.Since(tStreamPrep)
 
+	// A cached chunk whose volume server is down, or whose needle was evicted and
+	// now 404s under retry-backoff, would stall and then 500 after the headers are
+	// already sent -- serve the authoritative remote instead.
+	if s3a.shouldFallBackToRemote(entry, totalSize, versionId) {
+		if probeErr := probeReadable(ctx, reader, offset, localReadProbeTimeout); probeErr != nil {
+			if isCanceledStreamingError(probeErr) {
+				glog.V(3).Infof("streamFromVolumeServers: request canceled while probing %s/%s: %v", bucket, object, probeErr)
+				return probeErr
+			}
+			glog.V(1).Infof("streamFromVolumeServers: local read of %s/%s failed (%v), falling back to remote mount", bucket, object, probeErr)
+			if served, streamErr := s3a.serveObjectFromRemoteMount(w, r, entry, bucket, object, offset, size, isRangeRequest, totalSize, t0); served {
+				if streamErr != nil {
+					return newStreamErrorWithResponse(streamErr)
+				}
+				return nil
+			}
+			glog.Errorf("streamFromVolumeServers: local read of %s/%s failed and remote mount unavailable: %v", bucket, object, probeErr)
+			s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
+			return newStreamErrorWithResponse(probeErr)
+		}
+	}
+
 	// All validation and preparation successful - NOW set headers and write status
 	tHeaderSet := time.Now()
 	s3a.setResponseHeaders(w, r, entry, totalSize)
@@ -1170,40 +1183,147 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
 	// io.CopyBuffer drains them as fast memcpys.
 	tStreamExec := time.Now()
 	glog.V(4).Infof("streamFromVolumeServers: starting chunk reader, offset=%d, size=%d", offset, size)
-	cw := &countingWriter{w: w}
-	// Cap the copy buffer to the response size so small-object GETs (common
-	// for thumbnails, config files, etc.) don't allocate a 256 KiB scratch
-	// buffer per request.
-	const maxCopyBuf = 256 * 1024
-	copyBufSize := int64(maxCopyBuf)
-	if size > 0 && size < copyBufSize {
-		copyBufSize = size
-	}
-	copyBuf := make([]byte, copyBufSize)
-	_, err = io.CopyBuffer(cw, io.NewSectionReader(reader, offset, size), copyBuf)
+	written, err := s3a.streamRangeToClient(w, r, reader, entry, bucket, object, offset, size, totalSize, versionId)
 	streamExecTime = time.Since(tStreamExec)
 	// Track traffic even on partial writes for accurate egress accounting
-	if cw.written > 0 {
-		BucketTrafficSent(cw.written, r)
+	if written > 0 {
+		BucketTrafficSent(written, r)
 	}
 	if err != nil {
 		switch {
 		case isCanceledStreamingError(err):
 			// Client disconnected mid-stream (e.g. Nginx upstream timeout, browser cancel) - expected
-			glog.V(3).Infof("streamFromVolumeServers: client disconnected after writing %d bytes: %v", cw.written, err)
+			glog.V(3).Infof("streamFromVolumeServers: client disconnected after writing %d bytes: %v", written, err)
 		case errors.Is(err, context.DeadlineExceeded):
 			// Server-side deadline exceeded - unexpected, warrants operator attention
-			glog.Warningf("streamFromVolumeServers: server-side deadline exceeded after writing %d bytes: %v", cw.written, err)
+			glog.Warningf("streamFromVolumeServers: server-side deadline exceeded after writing %d bytes: %v", written, err)
 		default:
-			glog.Errorf("streamFromVolumeServers: streamFn failed after writing %d bytes: %v", cw.written, err)
+			glog.Errorf("streamFromVolumeServers: streamFn failed after writing %d bytes: %v", written, err)
 		}
 		// Streaming error after WriteHeader was called - response already partially written
 		return newStreamErrorWithResponse(err)
 	}
-	glog.V(4).Infof("streamFromVolumeServers: streamFn completed successfully, wrote %d bytes", cw.written)
+	glog.V(4).Infof("streamFromVolumeServers: streamFn completed successfully, wrote %d bytes", written)
 	return nil
 }
 
+// short, so a down volume server or an evicted needle under retry-backoff falls
+// back promptly instead of stalling the request
+const localReadProbeTimeout = 2 * time.Second
+
+// shouldFallBackToRemote reports whether a failed local read may be served from
+// the mounted remote instead: openRemoteStream only reads the unversioned key,
+// and the sizes must match for the bytes to be identical under the
+// already-computed Content-Length.
+func (s3a *S3ApiServer) shouldFallBackToRemote(entry *filer_pb.Entry, totalSize int64, versionId string) bool {
+	if v := resolvedSourceVersionId(versionId, entry); v != "" && v != "null" {
+		return false
+	}
+	remote := entry.GetRemoteEntry()
+	return totalSize > 0 && remote != nil && remote.GetRemoteSize() == totalSize
+}
+
+// the context-honoring read of filer.ChunkReadAt, so probeReadable can bound it
+type ctxReaderAt interface {
+	ReadAtWithTime(ctx context.Context, p []byte, offset int64) (int, int64, error)
+}
+
+// probeReadable decides whether the local copy is usable by reading one byte at
+// offset. Only a returned byte counts -- a zero-byte read, EOF or not, would
+// otherwise wave through a stream that then truncates.
+func probeReadable(ctx context.Context, reader ctxReaderAt, offset int64, timeout time.Duration) error {
+	probeCtx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+	var b [1]byte
+	n, _, err := reader.ReadAtWithTime(probeCtx, b[:], offset)
+	if n == len(b) {
+		return nil
+	}
+	if err == nil {
+		err = io.ErrUnexpectedEOF
+	}
+	return err
+}
+
+// serveObjectFromRemoteMount serves [offset, offset+size) straight from the
+// mounted remote. served=false means nothing was written and the caller still
+// owns the error path; once served is true the response is committed.
+func (s3a *S3ApiServer) serveObjectFromRemoteMount(w http.ResponseWriter, r *http.Request, entry *filer_pb.Entry, bucket, object string, offset, size int64, isRangeRequest bool, totalSize int64, t0 time.Time) (served bool, err error) {
+	remoteReader, remoteErr := s3a.openRemoteStream(r.Context(), bucket, object, offset, size, nil)
+	if remoteErr != nil {
+		glog.Warningf("streamFromVolumeServers: origin stream %s/%s: %v", bucket, object, remoteErr)
+		return false, nil
+	}
+	defer remoteReader.Close()
+
+	s3a.setResponseHeaders(w, r, entry, totalSize)
+	w.Header().Set("Content-Length", strconv.FormatInt(size, 10))
+	if isRangeRequest {
+		w.Header().Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", offset, offset+size-1, totalSize))
+		w.WriteHeader(http.StatusPartialContent)
+	} else {
+		w.WriteHeader(http.StatusOK)
+	}
+
+	TimeToFirstByte(r.Method, t0, r)
+	cw := &countingWriter{w: w}
+	_, copyErr := io.CopyN(cw, remoteReader, size)
+	if copyErr == io.EOF {
+		// the origin returned fewer bytes than the entry's RemoteSize
+		copyErr = io.ErrUnexpectedEOF
+	}
+	if cw.written > 0 {
+		BucketTrafficSent(cw.written, r)
+	}
+	if copyErr != nil {
+		glog.V(2).Infof("streamFromVolumeServers: origin stream %s/%s ended after %d bytes: %v", bucket, object, cw.written, copyErr)
+		return true, copyErr
+	}
+	return true, nil
+}
+
+// streamRangeToClient copies the local copy into the already-committed response.
+// The probe only proved the byte at offset readable, so a later chunk can still
+// be gone: finish such a body from the remote -- if it is still the generation
+// that was cached -- instead of truncating it under the Content-Length already
+// sent.
+func (s3a *S3ApiServer) streamRangeToClient(w io.Writer, r *http.Request, reader io.ReaderAt, entry *filer_pb.Entry, bucket, object string, offset, size, totalSize int64, versionId string) (int64, error) {
+	// small-object GETs shouldn't allocate a 256 KiB scratch buffer per request
+	const maxCopyBuf = 256 * 1024
+	copyBufSize := int64(maxCopyBuf)
+	if size > 0 && size < copyBufSize {
+		copyBufSize = size
+	}
+	cw := &countingWriter{w: w}
+	_, err := io.CopyBuffer(cw, io.NewSectionReader(reader, offset, size), make([]byte, copyBufSize))
+	if err == nil && cw.written < size {
+		// a short local read surfaces as a clean EOF
+		err = io.ErrUnexpectedEOF
+	}
+	if err == nil || cw.writeErr != nil || isCanceledStreamingError(err) || !s3a.shouldFallBackToRemote(entry, totalSize, versionId) {
+		return cw.written, err
+	}
+
+	remaining := size - cw.written
+	glog.V(1).Infof("streamFromVolumeServers: local read of %s/%s failed after %d bytes (%v), finishing from the remote mount", bucket, object, cw.written, err)
+	// the bytes already sent are the cached generation, so splice on the remote
+	// only while it still is that generation
+	remoteReader, remoteErr := s3a.openRemoteStream(r.Context(), bucket, object, offset+cw.written, remaining, entry.GetRemoteEntry())
+	if remoteErr != nil {
+		glog.Warningf("streamFromVolumeServers: origin stream %s/%s: %v", bucket, object, remoteErr)
+		return cw.written, err
+	}
+	defer remoteReader.Close()
+	if _, copyErr := io.CopyN(cw, remoteReader, remaining); copyErr != nil {
+		if copyErr == io.EOF {
+			copyErr = io.ErrUnexpectedEOF
+		}
+		glog.V(2).Infof("streamFromVolumeServers: origin stream %s/%s ended after %d bytes: %v", bucket, object, cw.written, copyErr)
+		return cw.written, copyErr
+	}
+	return cw.written, nil
+}
+
 // Shared HTTP client for volume server requests (connection pooling)
 var volumeServerHTTPClient = &http.Client{
 	Timeout: 5 * time.Minute,
@@ -3139,7 +3259,9 @@ func (s3a *S3ApiServer) buildRemoteObjectPath(bucket, object string) (dir, name
 
 // openRemoteStream opens a ranged read of a remote-only object straight from
 // its mounted origin, resolving the mount and storage conf from the filer.
-func (s3a *S3ApiServer) openRemoteStream(ctx context.Context, bucket, object string, offset, size int64) (io.ReadCloser, error) {
+// cached, when set, is the remote generation the local copy was made from: the
+// stream is refused unless the remote still matches it.
+func (s3a *S3ApiServer) openRemoteStream(ctx context.Context, bucket, object string, offset, size int64, cached *filer_pb.RemoteEntry) (io.ReadCloser, error) {
 	dir, name := s3a.buildRemoteObjectPath(bucket, object)
 
 	var storageConf *remote_pb.RemoteConf
@@ -3180,6 +3302,15 @@ func (s3a *S3ApiServer) openRemoteStream(ctx context.Context, bucket, object str
 	}
 
 	loc := filer.MapFullPathToRemoteStorageLocation(util.FullPath(localMountedDir), mountedLocation, util.FullPath(dir).Child(name))
+	if cached != nil {
+		current, statErr := client.StatFile(loc)
+		if statErr != nil {
+			return nil, statErr
+		}
+		if current.GetRemoteSize() != cached.GetRemoteSize() || current.GetRemoteETag() != cached.GetRemoteETag() || current.GetRemoteMtime() != cached.GetRemoteMtime() {
+			return nil, fmt.Errorf("remote %s/%s changed since it was cached", bucket, object)
+		}
+	}
 	return streamer.ReadFileAsStream(ctx, loc, offset, size)
 }
 

weed/s3api/s3api_remote_storage_test.go

@@ -609,11 +609,19 @@ func TestCacheRemoteObjectForStreamingCached(t *testing.T) {
 type fakeStreamRemoteClient struct {
 	remote_storage.RemoteStorageClient
 	data      []byte
+	stat      *filer_pb.RemoteEntry // what the remote reports now; defaults to matching data
 	gotLoc    *remote_pb.RemoteStorageLocation
 	gotOffset int64
 	gotSize   int64
 }
 
+func (c *fakeStreamRemoteClient) StatFile(loc *remote_pb.RemoteStorageLocation) (*filer_pb.RemoteEntry, error) {
+	if c.stat != nil {
+		return c.stat, nil
+	}
+	return &filer_pb.RemoteEntry{RemoteSize: int64(len(c.data))}, nil
+}
+
 func (c *fakeStreamRemoteClient) ReadFileAsStream(ctx context.Context, loc *remote_pb.RemoteStorageLocation, offset int64, size int64) (io.ReadCloser, error) {
 	c.gotLoc, c.gotOffset, c.gotSize = loc, offset, size
 	end := min(offset+size, int64(len(c.data)))