diff --git a/weed/command/mount_windows.go b/weed/command/mount_windows.go index 9cf4d0afff..71cd6a3189 100644 --- a/weed/command/mount_windows.go +++ b/weed/command/mount_windows.go @@ -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) diff --git a/weed/mount/dirty_pages_chunked.go b/weed/mount/dirty_pages_chunked.go index 469eb170b8..63852c2d8c 100644 --- a/weed/mount/dirty_pages_chunked.go +++ b/weed/mount/dirty_pages_chunked.go @@ -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() } diff --git a/weed/mount/dirty_pages_chunked_test.go b/weed/mount/dirty_pages_chunked_test.go index 7028f29532..b63715609a 100644 --- a/weed/mount/dirty_pages_chunked_test.go +++ b/weed/mount/dirty_pages_chunked_test.go @@ -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) + } +} diff --git a/weed/mount/weedfs_async_flush.go b/weed/mount/weedfs_async_flush.go index c7c3e83612..93031cb034 100644 --- a/weed/mount/weedfs_async_flush.go +++ b/weed/mount/weedfs_async_flush.go @@ -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. // diff --git a/weed/operation/upload_content.go b/weed/operation/upload_content.go index 71992dfe78..46d2f126ed 100644 --- a/weed/operation/upload_content.go +++ b/weed/operation/upload_content.go @@ -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) diff --git a/weed/operation/upload_content_test.go b/weed/operation/upload_content_test.go index 9e0a6e63a1..a5052fde5b 100644 --- a/weed/operation/upload_content_test.go +++ b/weed/operation/upload_content_test.go @@ -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) + } +}