feat: Support rapid appends for rw/w/w+ file open modes

docs/known-issues.md

@@ -13,7 +13,7 @@ This document lists known issues and bugs in GCSFuse, their impact, and the rele
 
 | Issue                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            | Affected Versions              | Fixed in                               | Reference                                                                                                                             |
 |:-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|:-------------------------------|:---------------------------------------|:--------------------------------------------------------------------------------------------------------------------------------------|
-| GCSFuse can lead to premature EOF on [A4X](https://docs.cloud.google.com/compute/docs/accelerator-optimized-machines#a4x-machine-type) and [A4X Max](https://docs.cloud.google.com/compute/docs/accelerator-optimized-machines#a4x-max-metal-machine-type) machines that feature a kernel with 64KiB page-size when setting a high [kernel-read-ahead](https://docs.cloud.google.com/storage/docs/cloud-storage-fuse/performance#increase-read-ahead-size) value.<br>**Workaround:** To fix this issue:<br>1. Either do not set the read-ahead<br>2. Or disable [streaming-writes](https://docs.cloud.google.com/storage/docs/cloud-storage-fuse/cli-options#enable-streaming-writes) by passing the flag `--enable-streaming-writes=false`. | \[v3.0.* - v3.7.1\]            | v3.7.2                                 ||
+| GCSFuse can lead to premature EOF/incorrect data reads on [A4X](https://docs.cloud.google.com/compute/docs/accelerator-optimized-machines#a4x-machine-type) and [A4X Max](https://docs.cloud.google.com/compute/docs/accelerator-optimized-machines#a4x-max-metal-machine-type) machines that feature a kernel with 64KiB page-size when setting a high [kernel-read-ahead](https://docs.cloud.google.com/storage/docs/cloud-storage-fuse/performance#increase-read-ahead-size) value.<br>**Workaround:** To fix this issue: Do not set the read-ahead | \[All versions up to v3.7.1\]            | v3.7.2                                 ||
 | Writing to a file fails with an Input/Output error on the application side, accompanied by 503 errors in the GCSFuse logs. This occurs when streaming writes are enabled. This issue was caused by stalls during write operations. It has been resolved in GCSFuse v3.3.0. Users should upgrade to v3.3.0 or later. **Workaround:** Disable streaming writes (`--enable-streaming-writes=false`) only if user can't upgrade. This flag reliably prevents the error only when staging writes uses fast media type like, SSD, tmpfs (specified using`--temp-dir`). | \[v3.0.* - v3.2.*\]            | v3.3.0                                 ||
 | Metrics: Input/output error when metrics are enabled. Applications may receive input/output errors from GCSFuse mounts when metrics are enabled.                                                                                                                                                                                                                                                                                                                                                                                                                 | v2.11.*                        | v2.12.0                                | [#3870](https://github.com/GoogleCloudPlatform/gcsfuse/issues/3870)                                                                   |
 | Metrics: Incorrect gcs/reader_count and gcs/download_bytes_count metrics.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        | \[v2.5.0, v3.4.*]              | v3.5.1                                 | [#3895](https://github.com/GoogleCloudPlatform/gcsfuse/pull/3895)                                                                     |

internal/fs/inode/file.go

@@ -1116,14 +1116,14 @@ func (f *FileInode) InitBufferedWriteHandlerIfEligible(ctx context.Context, open
 	var latestGcsObj *gcs.Object
 	var err error
 	if !f.local {
-		if f.bucket.BucketType().Zonal && openMode.IsAppend() {
+		if f.bucket.BucketType().Zonal && openMode.IsWrite() {
 			// In case of rapid appends, we will rely on kernel's latest view of the object
 			// instead of reaching out to the server for latest metadata. This is done to avoid
 			// forceful overwrites of local and latest object metadata with possibly stale server
 			// response. Since appends happen at the same generation, StatObject() call is redundant.
 			latestGcsObj = storageutil.ConvertMinObjectToObject(&f.src)
 		} else {
-			// For regional buckets or overwrites for rapid buckets, call StatObject() to fetch extended
+			// For regional buckets, call StatObject() to fetch extended
 			// attributes missing from the cached MinObject, which is required by the CreateObject request
 			// to create the new object generation.
 			latestGcsObj, err = f.fetchLatestGcsObject(ctx)
@@ -1169,7 +1169,7 @@ func (f *FileInode) areBufferedWritesSupported(openMode util.OpenMode, obj *gcs.
 	if f.local || obj.Size == 0 {
 		return true
 	}
-	if f.config.Write.EnableRapidAppends && openMode.IsAppend() && f.bucket.BucketType().Zonal && obj.Finalized.IsZero() {
+	if f.config.Write.EnableRapidAppends && openMode.IsWrite() && f.bucket.BucketType().Zonal && obj.Finalized.IsZero() {
 		return true
 	}
 	logger.Infof("Existing file %s of size %d bytes (non-zero) will use legacy staged writes. "+StreamingWritesSemantics, f.name.String(), obj.Size)

internal/fs/inode/file_mock_bucket_test.go

@@ -352,29 +352,7 @@ func (t *FileMockBucketTest) TestInitBufferedWriteHandlerIfEligible_ZonalBucket_
 	t.bucket.AssertNotCalled(t.T(), "StatObject", mock.Anything, mock.Anything)
 }
 
-func (t *FileMockBucketTest) TestInitBufferedWriteHandlerIfEligible_ZonalBucket_FetchesLatestMetadataFromGCS_ForOverwrites() {
-	// Setup Mock Bucket for Zonal
-	t.bucket = new(storagemock.TestifyMockBucket)
-	t.bucket.On("BucketType").Return(gcs.BucketType{Zonal: true})
-	// Setup expectations for inode creation
-	t.bucket.On("CreateObject", t.ctx, mock.AnythingOfType("*gcs.CreateObjectRequest")).
-		Return(&gcs.Object{Name: fileName, Size: 0, Generation: 1, MetaGeneration: 1}, nil)
-	// Create Inode (Non-local)
-	t.createLockedInode(fileName, emptyGCSFile)
-	t.in.content = nil // Force content to nil to allow BWH init
-	t.in.config.Write = *getWriteConfigWithEnabledRapidAppends()
-	// We expect to make a StatObject() call to GCS to fetch the latest minObject.
-	t.bucket.On("StatObject", t.ctx, mock.AnythingOfType("*gcs.StatObjectRequest")).
-		Return(&gcs.MinObject{Name: fileName, Size: 0, Generation: 1, MetaGeneration: 1}, &gcs.ExtendedObjectAttributes{}, nil)
-
-	// Call Method
-	initialized, err := t.in.InitBufferedWriteHandlerIfEligible(t.ctx, util.NewOpenMode(util.WriteOnly, 0))
 
-	// Assertions
-	require.NoError(t.T(), err)
-	assert.True(t.T(), initialized)
-	t.bucket.AssertExpectations(t.T())
-}
 
 func (t *FileMockBucketTest) TestInitBufferedWriteHandlerIfEligible_RegionalBucket_FetchesLatestMetadataFromGCS_Always() {
 	// Setup Mock Bucket for Non-Zonal

internal/util/file_util.go

@@ -60,6 +60,11 @@ func (om OpenMode) IsAppend() bool {
 	return om.accessMode != ReadOnly && om.fileFlags&O_APPEND != 0
 }
 
+// IsWrite returns true if the file was opened for writing.
+func (om OpenMode) IsWrite() bool {
+	return om.accessMode != ReadOnly
+}
+
 // IsDirect checks if the O_DIRECT flag is set, indicating that I/O should
 // bypass the kernel's page cache.
 func (om OpenMode) IsDirect() bool {

internal/util/file_util_test.go

@@ -15,6 +15,7 @@
 package util
 
 import (
+	"syscall"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -105,3 +106,42 @@ func TestFileOpenMode(t *testing.T) {
 		})
 	}
 }
+
+func TestIsWrite(t *testing.T) {
+	testCases := []struct {
+		name     string
+		openMode OpenMode
+		expected bool
+	}{
+		{
+			name:     "Read only",
+			openMode: NewOpenMode(ReadOnly, 0),
+			expected: false,
+		},
+		{
+			name:     "Write only",
+			openMode: NewOpenMode(WriteOnly, 0),
+			expected: true,
+		},
+		{
+			name:     "Read and write",
+			openMode: NewOpenMode(ReadWrite, 0),
+			expected: true,
+		},
+		{
+			name:     "Write only with append",
+			openMode: NewOpenMode(WriteOnly, syscall.O_APPEND),
+			expected: true,
+		},
+		{
+			name:     "Read and write with append",
+			openMode: NewOpenMode(ReadWrite, syscall.O_APPEND),
+			expected: true,
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			assert.Equal(t, tc.expected, tc.openMode.IsWrite())
+		})
+	}
+}

tools/integration_tests/rapid_appends/appends_test.go

@@ -15,6 +15,7 @@
 package rapid_appends
 
 import (
+	"io"
 	"os"
 	"path"
 	"syscall"
@@ -330,3 +331,35 @@ func TestDualMountAppendsTestSuite(t *testing.T) {
 		return &DualMountAppendsTestSuite{BaseSuite{primaryFlags: primaryFlags, secondaryFlags: secondaryFlags}}
 	})
 }
+
+func (t *SingleMountAppendsTestSuite) TestWriteAtEndToFinalizedObjectNotVisibleUntilClose() {
+	const initialContent = "dummy content"
+
+	t.fileName = fileNamePrefix + setup.GenerateRandomString(5)
+	// Create Finalized Object in the GCS bucket.
+	client.CreateFinalizedObjectInGCSTestDir(
+		testEnv.ctx, testEnv.storageClient, testDirName, t.fileName, initialContent, t.T())
+
+	// Append to the finalized object from the primary mount by seeking to the end.
+	data := setup.GenerateRandomString(contentSizeForBW * operations.OneMiB)
+	filePath := path.Join(t.primaryMount.testDirPath, t.fileName)
+	fh, err := os.OpenFile(filePath, os.O_WRONLY|syscall.O_DIRECT, operations.FilePermission_0600)
+	require.NoError(t.T(), err)
+	_, err = fh.Seek(0, io.SeekEnd)
+	require.NoError(t.T(), err)
+	n, err := fh.Write([]byte(data))
+	require.NoError(t.T(), err)
+	require.Equal(t.T(), len(data), n)
+
+	// Read from GCS to validate appended content is not yet visible.
+	contentBeforeClose, err := client.ReadObjectFromGCS(testEnv.ctx, testEnv.storageClient, path.Join(testDirName, t.fileName))
+	require.NoError(t.T(), err)
+	assert.Equal(t.T(), initialContent, string(contentBeforeClose))
+
+	// Close the file handle and verify appended content is now visible.
+	require.NoError(t.T(), fh.Close())
+	expectedContent := initialContent + data
+	contentAfterClose, err := client.ReadObjectFromGCS(testEnv.ctx, testEnv.storageClient, path.Join(testDirName, t.fileName))
+	require.NoError(t.T(), err)
+	assert.Equal(t.T(), expectedContent, string(contentAfterClose))
+}