mount: keep retrying writeback chunk uploads through a volume outage (#12465)

* mount: keep retrying writeback chunk uploads through a volume outage With -writebackCache, close() returns before the data is uploaded. When the volume servers stay unreachable past the upload's own retry (~20 s), the chunk was dropped and the file never appeared, with no error to the application. Retry transient upload errors with the same bytes until they succeed, unless the dirty pages were discarded. Permanent errors and the synchronous mode keep failing as before. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * mount: test writeback upload retries and the cases that still fail Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * operation: keep the filer's read-only verdict on assign failures UploadWithRetry flattened AssignVolumeResponse errors into a string, so callers could not tell a read-only rule from an outage. Wrap the filer's coded error so errors.Is(err, filer_pb.ErrReadOnly) works; the message is unchanged. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * mount: retry writeback uploads the upload layer reassigns, never read-only A filer proxy that cannot reach the volume answers 500 with an empty body, which util.IsTransientError does not match, so that outage still dropped the chunk. Also retry what operation.ShouldReassignUpload takes. A read-only rule stays final even when its path or message contains a transient word such as "unavailable". Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * mount: stop a writeback upload once its dirty pages are discarded Destroy() was only seen after a failed attempt, so a truncate during the retry wait still started another upload, and a chunk that finished after its pages were replaced was attached to the new content. Check before each attempt, and drop the chunk instead of attaching it; volume.fsck reclaims it. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * mount: keep Destroy from slipping between the discard check and the append truncateEntry destroys the pages and then installs the new entry, so a Destroy landing after the check but before AddChunks put the old chunk into the replacing entry. Hold one lock across the check and the append, and take it in Destroy while setting the flag. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * mount: leave the Windows interrupt drain to the capped filesystem hook Interrupt hooks run in reverse registration order, so the Windows hook ran before the filesystem hook and waited on pending writeback flushes without a bound. With writeback uploads now retrying through an outage, Ctrl-C could wait until the volume servers came back. Unmount only, as on Linux, and let the filesystem hook drain for at most 30 s and keep the write cache. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * mount: test a writeback upload whose assign keeps timing out With every volume server gone the assign RPC times out until the filer client gives up; that error must be retried like a refused upload. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-07 04:27:49 +01:00 · 2026-11-06 19:24:42 -08:00
1 parent 7d6680c7a4
commit fd81a50c57
@@ -123,12 +123,9 @@ func RunMount(option *MountOptions, umask os.FileMode) bool {
 	host.Notify(seaweedFileSystem)

 	grace.OnInterrupt(func() {
-		// The signal handler exits the process as soon as the hooks return, so
-		// anything still queued has to be flushed here rather than after Serve.
-		// WaitForAsyncFlush is idempotent, so the post-Serve call below is
-		// harmless if both run.
+		// The filesystem's own hook runs next and drains queued writeback
+		// flushes for at most 30 s, keeping the write cache if they are not done.
 		host.Unmount()
-		seaweedFileSystem.WaitForAsyncFlush()
 	})

 	serveMountGrpc(seaweedFileSystem, mountSocketListener)
@@ -1,13 +1,18 @@
 package mount

 import (
+	"errors"
 	"fmt"
 	"io"
 	"sync"
+	"sync/atomic"

+	"github.com/seaweedfs/seaweedfs/weed/filer"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/mount/page_writer"
+	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
 )

 type ChunkedDirtyPages struct {
@@ -19,16 +24,25 @@ type ChunkedDirtyPages struct {
 	replication    string
 	uploadPipeline *page_writer.UploadPipeline
 	hasWrites      bool
+
+	saveDataAsChunk func(util.FullPath) filer.SaveDataAsChunkFunctionType
+	destroyed       atomic.Bool
+	// attachLock keeps Destroy from landing between the destroyed check and
+	// the chunk append, which would put old bytes into the replacing entry.
+	attachLock sync.Mutex
 }

 var (
 	_ = page_writer.DirtyPages(&ChunkedDirtyPages{})
 )

+var errDirtyPagesDiscarded = errors.New("dirty pages discarded")
+
 func newMemoryChunkPages(fh *FileHandle, chunkSize int64) *ChunkedDirtyPages {

 	dirtyPages := &ChunkedDirtyPages{
-		fh: fh,
+		fh:              fh,
+		saveDataAsChunk: fh.wfs.saveDataAsChunk,
 	}

 	swapFileDir := fh.wfs.option.getUniqueCacheDirForWrite()
@@ -99,19 +113,61 @@ func (pages *ChunkedDirtyPages) saveChunkedFileIntervalToStorage(reader io.Reade

 	fileFullPath := pages.fh.FullPath()
 	fileName := fileFullPath.Name()
-	chunk, err := pages.fh.wfs.saveDataAsChunk(fileFullPath)(reader, fileName, offset, modifiedTsNs, uint64(size))
+	save := pages.saveDataAsChunk(fileFullPath)
+	var chunk *filer_pb.FileChunk
+	err := util.RetryUntil("writeback upload "+string(fileFullPath), func() (err error) {
+		if pages.destroyed.Load() {
+			return errDirtyPagesDiscarded
+		}
+		chunk, err = save(reader, fileName, offset, modifiedTsNs, uint64(size))
+		return err
+	}, func(err error) bool {
+		// close() already returned for a writeback handle, so nothing would
+		// see the error and the buffered data is the only copy left.
+		if !pages.fh.wfs.option.WritebackCache || !isRetryableUploadError(err) || pages.destroyed.Load() {
+			return false
+		}
+		seeker, ok := reader.(io.Seeker)
+		if !ok {
+			return false
+		}
+		if _, seekErr := seeker.Seek(0, io.SeekStart); seekErr != nil {
+			return false
+		}
+		glog.Warningf("%v writeback upload [%d,%d) failed, keeping the data and retrying: %v", fileFullPath, offset, offset+size, err)
+		return true
+	})
 	if err != nil {
 		glog.V(0).Infof("%v saveToStorage [%d,%d): %v", fileFullPath, offset, offset+size, err)
 		pages.setLastError(err)
 		return
 	}
+	// Truncation or newer content replaced these pages while they uploaded.
+	pages.attachLock.Lock()
+	defer pages.attachLock.Unlock()
+	if pages.destroyed.Load() {
+		glog.V(3).Infof("%v saveToStorage %s [%d,%d): pages discarded, chunk not attached", fileFullPath, chunk.FileId, offset, offset+size)
+		return
+	}
 	pages.fh.AddChunks([]*filer_pb.FileChunk{chunk})
 	pages.fh.entryChunkGroup.AddChunk(chunk)
 	glog.V(3).Infof("%v saveToStorage %s [%d,%d)", fileFullPath, chunk.FileId, offset, offset+size)

 }

+// isRetryableUploadError also takes the failures the upload layer reassigns on,
+// such as a filer proxy answering 500, but never a read-only verdict.
+func isRetryableUploadError(err error) bool {
+	if errors.Is(err, filer_pb.ErrReadOnly) {
+		return false
+	}
+	return util.IsTransientError(err) || operation.ShouldReassignUpload(err)
+}
+
 func (pages *ChunkedDirtyPages) Destroy() {
+	pages.attachLock.Lock()
+	pages.destroyed.Store(true)
+	pages.attachLock.Unlock()
 	pages.uploadPipeline.Shutdown()
 }

@@ -1,8 +1,23 @@
 package mount

 import (
+	"context"
 	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"syscall"
 	"testing"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/operation"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/util"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )

 // Once a chunk upload fails its data is gone, so the handle must reject
@@ -29,3 +44,146 @@ func TestChunkedDirtyPagesFailWritesAfterUploadError(t *testing.T) {
 		t.Fatalf("LastError = %v, want first error %v", err, uploadErr)
 	}
 }
+
+type flakySave struct {
+	failures []error
+	calls    int
+	bodies   []string
+	onCall   func()
+}
+
+func (s *flakySave) saveDataAsChunk(util.FullPath) filer.SaveDataAsChunkFunctionType {
+	return func(reader io.Reader, name string, offset int64, tsNs int64, _ uint64) (*filer_pb.FileChunk, error) {
+		body, _ := io.ReadAll(reader)
+		s.bodies = append(s.bodies, string(body))
+		s.calls++
+		if s.onCall != nil {
+			s.onCall()
+		}
+		if s.calls <= len(s.failures) {
+			return nil, s.failures[s.calls-1]
+		}
+		return &filer_pb.FileChunk{FileId: "1,01", Offset: offset, Size: uint64(len(body)), ModifiedTsNs: tsNs}, nil
+	}
+}
+
+func newUploadTestPages(t *testing.T, writeback bool, save *flakySave) *ChunkedDirtyPages {
+	t.Helper()
+	wfs := newInvalidateTestWFS(t)
+	wfs.option.WritebackCache = writeback
+	fh := newFileHandle(wfs, 1, 2, &filer_pb.Entry{Name: "f", Attributes: &filer_pb.FuseAttributes{}})
+	fh.RememberPath("/dir/f")
+	pages := newMemoryChunkPages(fh, 1024)
+	t.Cleanup(pages.Destroy)
+	pages.saveDataAsChunk = save.saveDataAsChunk
+	return pages
+}
+
+var errVolumeRefused = fmt.Errorf("upload data: %w", &net.OpError{Op: "dial", Net: "tcp", Err: syscall.ECONNREFUSED})
+
+// filerProxyOutage is what a filer proxy sends when it cannot reach the
+// volume server: a 500 with an empty body.
+func filerProxyOutage(t *testing.T) error {
+	t.Helper()
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer server.Close()
+	_, err := operation.NewUploaderWithHttpClient(server.Client()).UploadData(context.Background(), []byte("hello"),
+		&operation.UploadOption{UploadUrl: server.URL, MaxAttempts: 1})
+	if err == nil {
+		t.Fatal("upload to the failing proxy succeeded")
+	}
+	return fmt.Errorf("upload data: %w", err)
+}
+
+// close() has already returned for a writeback handle, so a volume outage must
+// not drop the buffered chunk: the same bytes go out again once it ends.
+func TestChunkedDirtyPagesWritebackRetriesTransientUploadError(t *testing.T) {
+	for name, outage := range map[string]error{
+		"volume refused":  errVolumeRefused,
+		"filer proxy 500": filerProxyOutage(t),
+		// With no volume server left the assign keeps timing out, and the
+		// filer client gives up after its own retries.
+		"assign timed out": fmt.Errorf("upload data: filerGrpcAddress assign volume: %w",
+			status.Error(codes.Unavailable, "filer 127.0.0.1:18888: call timed out")),
+	} {
+		t.Run(name, func(t *testing.T) {
+			save := &flakySave{failures: []error{outage}}
+			pages := newUploadTestPages(t, true, save)
+
+			pages.saveChunkedFileIntervalToStorage(util.NewBytesReader([]byte("hello")), 0, 5, 1, func() {})
+
+			if err := pages.LastError(); err != nil {
+				t.Fatalf("LastError = %v, want the retried upload to succeed", err)
+			}
+			if save.calls != 2 || save.bodies[1] != "hello" {
+				t.Fatalf("uploads = %d with bodies %q, want a second upload of the same bytes", save.calls, save.bodies)
+			}
+			if chunks := pages.fh.GetEntry().GetChunks(); len(chunks) != 1 || chunks[0].Size != 5 {
+				t.Fatalf("entry chunks = %v, want the retried chunk", chunks)
+			}
+		})
+	}
+}
+
+func TestChunkedDirtyPagesUploadErrorWithoutRetry(t *testing.T) {
+	permanent := errors.New("upload data: assign volume failure: no writable volumes")
+	readOnly := fmt.Errorf("upload data: assign volume failure path:\"/unavailable/f\": %w",
+		filer_pb.AssignVolumeResponseError(&filer_pb.AssignVolumeResponse{
+			Error:     "assign volume: read only: /unavailable",
+			ErrorCode: filer_pb.FilerError_READ_ONLY,
+		}))
+	tests := []struct {
+		name      string
+		writeback bool
+		err       error
+		discard   func(pages *ChunkedDirtyPages)
+		wantErr   error
+	}{
+		{name: "sync mode reports the outage to close", err: errVolumeRefused},
+		{name: "writeback permanent error", writeback: true, err: permanent},
+		{name: "writeback read-only path", writeback: true, err: readOnly},
+		{name: "writeback pages discarded mid-upload", writeback: true, err: errVolumeRefused,
+			discard: func(pages *ChunkedDirtyPages) { pages.Destroy() }},
+		{name: "writeback pages discarded during the retry wait", writeback: true, err: errVolumeRefused,
+			discard: func(pages *ChunkedDirtyPages) { time.AfterFunc(100*time.Millisecond, pages.Destroy) },
+			wantErr: errDirtyPagesDiscarded},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			save := &flakySave{failures: []error{tt.err}}
+			pages := newUploadTestPages(t, tt.writeback, save)
+			if tt.discard != nil {
+				save.onCall = func() { tt.discard(pages) }
+			}
+			wantErr := tt.err
+			if tt.wantErr != nil {
+				wantErr = tt.wantErr
+			}
+
+			pages.saveChunkedFileIntervalToStorage(util.NewBytesReader([]byte("hello")), 0, 5, 1, func() {})
+
+			if save.calls != 1 {
+				t.Fatalf("uploads = %d, want 1", save.calls)
+			}
+			if err := pages.LastError(); !errors.Is(err, wantErr) {
+				t.Fatalf("LastError = %v, want %v", err, wantErr)
+			}
+		})
+	}
+}
+
+// A chunk whose pages were discarded while it uploaded must not reappear in
+// the handle's entry, e.g. after a truncate replaced the content.
+func TestChunkedDirtyPagesDiscardedUploadNotAttached(t *testing.T) {
+	save := &flakySave{}
+	pages := newUploadTestPages(t, true, save)
+	save.onCall = pages.Destroy
+
+	pages.saveChunkedFileIntervalToStorage(util.NewBytesReader([]byte("hello")), 0, 5, 1, func() {})
+
+	if chunks := pages.fh.GetEntry().GetChunks(); len(chunks) != 0 {
+		t.Fatalf("entry chunks = %v, want none from discarded pages", chunks)
+	}
+}
@@ -56,12 +56,11 @@ func (wfs *WFS) completeAsyncFlush(fh *FileHandle) {
 		fh.inode, fh.fh, fh.savedDir, fh.savedName, fh.dirtyMetadata, fh.isDeleted, fh.isRenamed)

 	// Phase 1: Flush dirty pages — seals writable chunks, uploads to volume servers, and waits.
-	// The underlying UploadWithRetry already retries transient HTTP/gRPC errors internally,
-	// so a failure here indicates a persistent issue; the chunk data has been freed.
+	// Writeback uploads retry transient errors until they succeed, so a failure here
+	// is a permanent one and the chunk data has been freed.
 	if err := fh.dirtyPages.FlushData(); err != nil {
 		glog.Errorf("completeAsyncFlush inode %d: data flush failed: %v", fh.inode, err)
-		// Data is lost at this point (chunks freed after internal retry exhaustion).
-		// Proceed to cleanup to avoid resource leaks and unmount hangs.
+		// Data is lost at this point. Proceed to cleanup to avoid resource leaks and unmount hangs.
 	} else if fh.dirtyMetadata {
 		// Phase 2: Flush metadata unless the file was explicitly unlinked.
 		//
@@ -295,7 +295,7 @@ func (uploader *Uploader) UploadWithRetry(filerClient filer_pb.FilerClient, assi
 				return assignErr
 			}
 			if resp.Error != "" {
-				return fmt.Errorf("assign volume failure %v: %v", assignRequest, resp.Error)
+				return fmt.Errorf("assign volume failure %v: %w", assignRequest, filer_pb.AssignVolumeResponseError(resp))
 			}

 			fileId, auth = resp.FileId, security.EncodedJwt(resp.Auth)
@@ -990,3 +990,17 @@ func TestCipherUploadKeepsPlaintextETag(t *testing.T) {
 		t.Fatalf("decrypted %d bytes, want %d", len(decrypted), len(payload))
 	}
 }
+
+// A read-only rule is the filer's verdict, not an outage: callers that retry
+// transient failures must still be able to tell it apart.
+func TestUploadWithRetryKeepsReadOnlyVerdict(t *testing.T) {
+	client := &fsyncAssignClient{response: &filer_pb.AssignVolumeResponse{
+		Error:     "assign volume: read only: /unavailable",
+		ErrorCode: filer_pb.FilerError_READ_ONLY,
+	}}
+	_, _, err, _ := newUploader(http.DefaultClient).UploadWithRetry(client,
+		&filer_pb.AssignVolumeRequest{Path: "/unavailable/object"}, &UploadOption{}, strings.NewReader("abc"))
+	if !errors.Is(err, filer_pb.ErrReadOnly) {
+		t.Fatalf("UploadWithRetry error = %v, want it to wrap filer_pb.ErrReadOnly", err)
+	}
+}