Doing flush for pending writes on rename file

internal/fs/fs.go

@@ -2052,10 +2052,57 @@ func (fs *fileSystem) Rename(
 		}
 		return fs.renameNonHierarchicalDir(ctx, oldParent, op.OldName, newParent, op.NewName)
 	}
+
+	return fs.renameFile(ctx, op, child, oldParent, newParent)
+}
+
+// LOCKS_EXCLUDED(oldParent)
+// LOCKS_EXCLUDED(newParent)
+func (fs *fileSystem) renameFile(ctx context.Context, op *fuseops.RenameOp, child *inode.Core, oldParent inode.DirInode, newParent inode.DirInode) error {
+	updatedMinObject, err := fs.flushBeforeRename(ctx, child)
+	if err != nil {
+		return fmt.Errorf("flushBeforeRename error :%v", err)
+	}
+
 	if (child.Bucket.BucketType().Hierarchical && fs.enableAtomicRenameObject) || child.Bucket.BucketType().Zonal {
-		return fs.renameHierarchicalFile(ctx, oldParent, op.OldName, child.MinObject, newParent, op.NewName)
+		return fs.renameHierarchicalFile(ctx, oldParent, op.OldName, updatedMinObject, newParent, op.NewName)
+	}
+	return fs.renameNonHierarchicalFile(ctx, oldParent, op.OldName, updatedMinObject, newParent, op.NewName)
+}
+
+// LOCKS_EXCLUDED(fs.mu)
+// LOCKS_EXCLUDED(fileInode)
+func (fs *fileSystem) flushBeforeRename(ctx context.Context, child *inode.Core) (minObject *gcs.MinObject, err error) {
+	// We will return modified minObject if flush is done, otherwise the original
+	// minObject is returned. Original minObject is the one passed in the request.
+	minObject = child.MinObject
+	if !fs.newConfig.Write.EnableStreamingWrites {
+		return
+	}
+
+	fs.mu.Lock()
+	existingInode, ok := fs.generationBackedInodes[child.FullName]
+	fs.mu.Unlock()
+	// If this is not an existing file, then nothing to do.
+	// We shouldn't hit scenario ideally since we checked for local file and
+	// explicit-dirs (flat bucket) before reaching this step.
+	if !ok {
+		logger.Warnf("Encountered a non-fileInode in rename file path %d", existingInode.ID())
+		return
+	}
+
+	fileInode, isFileInode := existingInode.(*inode.FileInode)
+	// Same as above comment. This should be a file for sure.
+	if !isFileInode {
+		logger.Warnf("Encountered a non-fileInode in rename file path %d", existingInode.ID())
+		return
 	}
-	return fs.renameNonHierarchicalFile(ctx, oldParent, op.OldName, child.MinObject, newParent, op.NewName)
+
+	fileInode.Lock()
+	defer fileInode.Unlock()
+	// Try to flush if there are any pending writes.
+	err = fs.flushFile(ctx, fileInode)
+	return
 }
 
 // LOCKS_EXCLUDED(oldParent)

tools/integration_tests/streaming_writes/move_file_test.go

@@ -0,0 +1,43 @@
+// Copyright 2025 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package streaming_writes
+
+import (
+	"path"
+
+	. "github.com/googlecloudplatform/gcsfuse/v2/tools/integration_tests/util/client"
+	"github.com/googlecloudplatform/gcsfuse/v2/tools/integration_tests/util/operations"
+	"github.com/stretchr/testify/require"
+)
+
+func (t *defaultMountEmptyGCSFile) TestMoveBeforeFileIsFlushed() {
+	operations.WriteWithoutClose(t.f1, FileContents, t.T())
+	operations.WriteWithoutClose(t.f1, FileContents, t.T())
+	operations.VerifyStatFile(t.filePath, int64(2*len(FileContents)), FilePerms, t.T())
+	err := t.f1.Sync()
+	require.NoError(t.T(), err)
+
+	newFile := "newFile.txt"
+	destDirPath := path.Join(testDirPath, newFile)
+	err = operations.Move(t.filePath, destDirPath)
+
+	// Validate that move didn't throw any error.
+	require.NoError(t.T(), err)
+	// Verify the new object contents.
+	ValidateObjectContentsFromGCS(ctx, storageClient, testDirName, newFile, FileContents+FileContents, t.T())
+	require.NoError(t.T(), t.f1.Close())
+	// Check if old object is deleted.
+	ValidateObjectNotFoundErrOnGCS(ctx, storageClient, testDirName, t.fileName, t.T())
+}