diff --git a/test/sftp/basic_test.go b/test/sftp/basic_test.go index bad875208..17f3b191f 100644 --- a/test/sftp/basic_test.go +++ b/test/sftp/basic_test.go @@ -3,6 +3,7 @@ package sftp import ( "bytes" "io" + "os" "path" "testing" @@ -602,6 +603,32 @@ func TestFileContent(t *testing.T) { sftpClient.Remove("/binary.bin") }) + t.Run("WriteAtOffsetKeepsExistingBytes", func(t *testing.T) { + file, err := sftpClient.Create("/resume.txt") + require.NoError(t, err) + _, err = file.Write([]byte("hello world")) + require.NoError(t, err) + require.NoError(t, file.Close()) + + // Reopen without O_TRUNC and write past the start, as a resumed upload does + file, err = sftpClient.OpenFile("/resume.txt", os.O_WRONLY) + require.NoError(t, err) + _, err = file.WriteAt([]byte("WORLD!"), 6) + require.NoError(t, err) + require.NoError(t, file.Close()) + + readFile, err := sftpClient.Open("/resume.txt") + require.NoError(t, err) + content, err := io.ReadAll(readFile) + require.NoError(t, err) + readFile.Close() + + require.Equal(t, "hello WORLD!", string(content)) + + // Clean up + sftpClient.Remove("/resume.txt") + }) + t.Run("EmptyFile", func(t *testing.T) { file, err := sftpClient.Create("/empty.txt") require.NoError(t, err) diff --git a/weed/filer/filechunk_manifest.go b/weed/filer/filechunk_manifest.go index a8661549c..3eacae5cd 100644 --- a/weed/filer/filechunk_manifest.go +++ b/weed/filer/filechunk_manifest.go @@ -458,13 +458,13 @@ func retriedStreamFetchChunkData(ctx context.Context, writer io.Writer, urlStrin localProcessed += writtenCount totalWritten += writtenCount }) - if !shouldRetry { - break - } if writeErr != nil { err = writeErr break } + if !shouldRetry { + break + } if err != nil { failed = true glog.V(0).InfofCtx(ctx, "read %s failed, err: %v", urlString, err) diff --git a/weed/filer/filechunk_manifest_test.go b/weed/filer/filechunk_manifest_test.go index e23bf5366..5e82c7045 100644 --- a/weed/filer/filechunk_manifest_test.go +++ b/weed/filer/filechunk_manifest_test.go @@ -787,3 +787,23 @@ func TestFetchWholeChunkUsesProvidedFilerJwt(t *testing.T) { return "" })) } + +// A completed fetch whose destination write failed is a failed copy, not a +// short success: the caller must see the write error, or a temp disk filling +// mid-copy would still upload the truncated file. +func TestRetriedStreamFetchChunkDataReportsWriteError(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte("0123456789")) + })) + t.Cleanup(srv.Close) + + written, err := retriedStreamFetchChunkData(context.Background(), &failOnWrite{}, []string{srv.URL + "/5,abc"}, "", nil, false, true, 0, 0, nil) + assert.Error(t, err) + assert.Zero(t, written) +} + +type failOnWrite struct{} + +func (w *failOnWrite) Write(p []byte) (int, error) { + return 0, fmt.Errorf("no space left") +} diff --git a/weed/sftpd/sftp_copy_existing_test.go b/weed/sftpd/sftp_copy_existing_test.go new file mode 100644 index 000000000..5f603107a --- /dev/null +++ b/weed/sftpd/sftp_copy_existing_test.go @@ -0,0 +1,210 @@ +package sftpd + +import ( + "bytes" + "context" + "net" + "net/http" + "net/http/httptest" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/sftpd/user" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" +) + +type copyTestFiler struct { + filer_pb.UnimplementedSeaweedFilerServer + entries map[string]*filer_pb.Entry + volumeUrl string +} + +func (f *copyTestFiler) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { + if entry, ok := f.entries[req.Directory+"/"+req.Name]; ok { + return &filer_pb.LookupDirectoryEntryResponse{Entry: entry}, nil + } + return nil, status.Error(codes.NotFound, "no entry is found") +} + +func (f *copyTestFiler) LookupVolume(_ context.Context, req *filer_pb.LookupVolumeRequest) (*filer_pb.LookupVolumeResponse, error) { + resp := &filer_pb.LookupVolumeResponse{LocationsMap: make(map[string]*filer_pb.Locations)} + for _, vid := range req.VolumeIds { + resp.LocationsMap[vid] = &filer_pb.Locations{Locations: []*filer_pb.Location{{Url: f.volumeUrl}}} + } + return resp, nil +} + +func newCopyTestServer(t *testing.T, volumeUrl string, entries map[string]*filer_pb.Entry) *SftpServer { + lis, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + grpcServer := grpc.NewServer() + filer_pb.RegisterSeaweedFilerServer(grpcServer, ©TestFiler{entries: entries, volumeUrl: volumeUrl}) + go func() { _ = grpcServer.Serve(lis) }() + t.Cleanup(grpcServer.Stop) + + _, port, _ := net.SplitHostPort(lis.Addr().String()) + return &SftpServer{ + filerAddr: pb.ServerAddress("127.0.0.1:0." + port), + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + user: &user.User{Username: "alice", Uid: 1000, HomeDir: "/"}, + } +} + +func chunkedEntry(size int64, chunks ...*filer_pb.FileChunk) *filer_pb.Entry { + return &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{FileSize: uint64(size)}, + Chunks: chunks, + IsDirectory: false, + } +} + +func fileChunk(fileId string, offset, size int64) *filer_pb.FileChunk { + return &filer_pb.FileChunk{FileId: fileId, Offset: offset, Size: uint64(size), ModifiedTsNs: 1} +} + +// A non-truncating write must keep untouched bytes, including a sparse file's +// zeros between chunks. Stopping at the first gap would drop the tail. +func TestCopyExistingContentFillsSparseGaps(t *testing.T) { + chunkA := []byte("AAAAAAAAAA") + chunkB := []byte("BBBBBBBBBB") + objects := map[string][]byte{"3,01": chunkA, "3,02": chunkB} + volume := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + data, ok := objects[r.URL.Path[1:]] + if !ok { + http.Error(w, "not found", http.StatusNotFound) + return + } + _, _ = w.Write(data) + })) + t.Cleanup(volume.Close) + + entries := map[string]*filer_pb.Entry{ + "/home/sparse.bin": chunkedEntry(100, fileChunk("3,01", 0, 10), fileChunk("3,02", 90, 10)), + } + fs := newCopyTestServer(t, volume.Listener.Addr().String(), entries) + + var dst bytes.Buffer + if err := fs.copyExistingContent("/home/sparse.bin", &dst); err != nil { + t.Fatalf("copyExistingContent: %v", err) + } + want := append(append(append([]byte{}, chunkA...), make([]byte, 80)...), chunkB...) + if !bytes.Equal(dst.Bytes(), want) { + t.Fatalf("copied %d bytes, want %d with a zero-filled gap; tail present: %v", dst.Len(), len(want), bytes.Contains(dst.Bytes(), chunkB)) + } +} + +// A remote-only file has no local chunks: copying nothing would upload an +// empty replacement on close. Refusing the read keeps the remote object. +func TestCopyExistingContentRejectsUnreadableEntry(t *testing.T) { + remoteOnly := chunkedEntry(0) + remoteOnly.RemoteEntry = &filer_pb.RemoteEntry{RemoteSize: 100} + entries := map[string]*filer_pb.Entry{ + "/home/remote.bin": remoteOnly, + } + fs := newCopyTestServer(t, "", entries) + + var dst bytes.Buffer + if err := fs.copyExistingContent("/home/remote.bin", &dst); err == nil { + t.Fatal("expected refusal when no local bytes exist") + } +} + +// A locally shortened file still names its older, larger remote object. Once +// the local entry is newer the remote size is stale: copying must not pad the +// file back out to it. +func TestCopyExistingContentIgnoresStaleRemoteSize(t *testing.T) { + chunkA := []byte("AAAAAAAAAA") + volume := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/3,01" { + http.Error(w, "not found", http.StatusNotFound) + return + } + _, _ = w.Write(chunkA) + })) + t.Cleanup(volume.Close) + + entries := map[string]*filer_pb.Entry{ + "/home/short.bin": { + Attributes: &filer_pb.FuseAttributes{FileSize: 10, Mtime: 200}, + Chunks: []*filer_pb.FileChunk{fileChunk("3,01", 0, 10)}, + RemoteEntry: &filer_pb.RemoteEntry{RemoteSize: 100, RemoteMtime: 100}, + }, + } + fs := newCopyTestServer(t, volume.Listener.Addr().String(), entries) + + var dst bytes.Buffer + if err := fs.copyExistingContent("/home/short.bin", &dst); err != nil { + t.Fatalf("copyExistingContent: %v", err) + } + if !bytes.Equal(dst.Bytes(), chunkA) { + t.Fatalf("copied %d bytes, want the %d local bytes only", dst.Len(), len(chunkA)) + } +} + +// A file whose size is all holes still keeps its zeros. +func TestCopyExistingContentAllHoles(t *testing.T) { + entries := map[string]*filer_pb.Entry{ + "/home/holes.bin": chunkedEntry(10), + } + fs := newCopyTestServer(t, "", entries) + + var dst bytes.Buffer + if err := fs.copyExistingContent("/home/holes.bin", &dst); err != nil { + t.Fatalf("all-holes file: %v", err) + } + if dst.Len() != 10 || !bytes.Equal(dst.Bytes(), make([]byte, 10)) { + t.Fatalf("wrote %d bytes, want 10 zeros", dst.Len()) + } +} + +func TestCopyExistingContentInlineAndMissing(t *testing.T) { + entries := map[string]*filer_pb.Entry{ + "/home/inline.txt": {Attributes: &filer_pb.FuseAttributes{FileSize: 5}, Content: []byte("hello")}, + } + fs := newCopyTestServer(t, "", entries) + + var dst bytes.Buffer + if err := fs.copyExistingContent("/home/inline.txt", &dst); err != nil { + t.Fatalf("inline content: %v", err) + } + if got := dst.String(); got != "hello" { + t.Fatalf("inline content = %q, want %q", got, "hello") + } + + dst.Reset() + if err := fs.copyExistingContent("/home/missing.txt", &dst); err != nil || dst.Len() != 0 { + t.Fatalf("missing file: err=%v written=%d", err, dst.Len()) + } +} + +func TestCopyExistingContentTrailingGap(t *testing.T) { + chunkA := []byte("AAAAAAAAAA") + volume := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/3,01" { + http.Error(w, "not found", http.StatusNotFound) + return + } + _, _ = w.Write(chunkA) + })) + t.Cleanup(volume.Close) + + entries := map[string]*filer_pb.Entry{ + "/home/trailing.bin": chunkedEntry(50, fileChunk("3,01", 0, 10)), + } + fs := newCopyTestServer(t, volume.Listener.Addr().String(), entries) + + var dst bytes.Buffer + if err := fs.copyExistingContent("/home/trailing.bin", &dst); err != nil { + t.Fatalf("copyExistingContent: %v", err) + } + want := append(append([]byte{}, chunkA...), make([]byte, 40)...) + if !bytes.Equal(dst.Bytes(), want) { + t.Fatalf("copied %d bytes, want %d with a zero-filled tail", dst.Len(), len(want)) + } +} diff --git a/weed/sftpd/sftp_filer.go b/weed/sftpd/sftp_filer.go index eb4a989a6..1262b75d8 100644 --- a/weed/sftpd/sftp_filer.go +++ b/weed/sftpd/sftp_filer.go @@ -15,6 +15,7 @@ import ( "time" "github.com/pkg/sftp" + "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/pb" filer_pb "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -23,6 +24,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/sftpd/user" "github.com/seaweedfs/seaweedfs/weed/util" util_http "github.com/seaweedfs/seaweedfs/weed/util/http" + "github.com/seaweedfs/seaweedfs/weed/wdclient" "google.golang.org/grpc" ) @@ -159,6 +161,14 @@ func (fs *SftpServer) newFileWriter(r *sftp.Request) (io.WriterAt, error) { if err != nil { return nil, fmt.Errorf("failed to create temp file: %w", err) } + // Without O_TRUNC the client writes into the existing file, so start from its current bytes. + if !r.Pflags().Trunc { + if err := fs.copyExistingContent(absPath, tmpFile); err != nil { + tmpFile.Close() + os.Remove(tmpFile.Name()) + return nil, err + } + } return &SeaweedSftpFileWriter{ fs: *fs, @@ -172,6 +182,75 @@ func (fs *SftpServer) newFileWriter(r *sftp.Request) (io.WriterAt, error) { }, nil } +func (fs *SftpServer) copyExistingContent(absPath string, dst io.Writer) error { + entry, err := fs.getEntry(absPath) + if err == os.ErrNotExist { + return nil + } + if err != nil { + return err + } + if entry.IsDirectory { + return nil + } + size := int64(filer.FileSize(entry)) + remote := entry.GetRemoteEntry() + if len(entry.Content) == 0 && len(entry.GetChunks()) == 0 { + if remote != nil && remote.RemoteSize > 0 { + return fmt.Errorf("existing content of %s is not readable", absPath) + } + if size == 0 { + return nil + } + return writeZeros(dst, size) + } + if size == 0 { + return nil + } + if len(entry.Content) > 0 { + if _, err := dst.Write(entry.Content); err != nil { + return err + } + if remaining := size - int64(len(entry.Content)); remaining > 0 { + return writeZeros(dst, remaining) + } + return nil + } + streamFn, err := filer.PrepareStreamContent(lookupFileIdFn(filer.LookupFn(fs)), filer.JwtForVolumeServer, entry.GetChunks(), 0, size) + if err != nil { + return fmt.Errorf("read existing content of %s: %w", absPath, err) + } + if err := streamFn(dst); err != nil { + return fmt.Errorf("read existing content of %s: %w", absPath, err) + } + return nil +} + +// lookupFileIdFn adapts a filer LookupFn to the chunk streamer's provider +// interface. PrepareStreamContent fills unstored gaps with zeros, so a sparse +// file's tail survives the copy. +type lookupFileIdFn wdclient.LookupFileIdFunctionType + +func (fn lookupFileIdFn) GetLookupFileIdFunction() wdclient.LookupFileIdFunctionType { + return wdclient.LookupFileIdFunctionType(fn) +} + +func writeZeros(w io.Writer, size int64) error { + zeros := make([]byte, 8192) + for size > 0 { + n := int64(len(zeros)) + if size < n { + n = size + } + written, err := w.Write(zeros[:n]) + size -= int64(written) + if err != nil { + return err + } + } + return nil +} + func (fs *SftpServer) removeEntry(absPath string) error { return fs.deleteEntry(absPath, false) }